feature: Errors page (#3172)

A top-level Errors page that aggregates errors from failed runs with
occurrences metrics.


https://github.com/user-attachments/assets/8f0ef55e-90dd-4faa-9051-59f4665181e4

Errors are “fingerprinted” so similar errors are grouped together (e.g.
has an ID in the error message).

You can view an individual error to view a timeline of when it fired,
the runs, and bulk replay them.
This commit is contained in:
Matt Aitken
2026-03-05 11:05:45 +00:00
committed by GitHub
parent e49ccc1226
commit 5f359be286
38 changed files with 4284 additions and 15 deletions
+11
View File
@@ -0,0 +1,11 @@
---
area: webapp
type: feature
---
A new Errors page for viewing and tracking errors that cause runs to fail
- Errors are grouped using error fingerprinting
- View top errors for a time period, filter by task, or search the text
- View occurrences over time
- View all the runs for an error and bulk replay them
@@ -228,6 +228,18 @@ export function BulkActionFilterSummary({
/>
);
}
case "errorId": {
return (
<AppliedFilter
variant="minimal/medium"
key={key}
label={"Error ID"}
icon={filterIcon(key)}
value={value}
removable={false}
/>
);
}
default: {
assertNever(typedKey);
}
@@ -3,12 +3,14 @@ import { motion } from "framer-motion";
import { useCallback, useEffect, useRef, useState } from "react";
import { Input } from "~/components/primitives/Input";
import { ShortcutKey } from "~/components/primitives/ShortcutKey";
import { cn } from "~/utils/cn";
import { useOptimisticLocation } from "~/hooks/useOptimisticLocation";
import { useSearchParams } from "~/hooks/useSearchParam";
import { cn } from "~/utils/cn";
export function LogsSearchInput() {
const location = useOptimisticLocation();
export type LogsSearchInputProps = {
placeholder?: string;
};
export function LogsSearchInput({ placeholder = "Search logs…" }: LogsSearchInputProps) {
const inputRef = useRef<HTMLInputElement>(null);
const { value, replace, del } = useSearchParams();
@@ -61,7 +63,7 @@ export function LogsSearchInput() {
type="text"
ref={inputRef}
variant="secondary-small"
placeholder="Search logs…"
placeholder={placeholder}
value={text}
onChange={(e) => setText(e.target.value)}
fullWidth
@@ -24,6 +24,7 @@ import {
Squares2X2Icon,
TableCellsIcon,
UsersIcon,
BugAntIcon,
} from "@heroicons/react/20/solid";
import { Link, useFetcher, useNavigation } from "@remix-run/react";
import { LayoutGroup, motion } from "framer-motion";
@@ -73,6 +74,7 @@ import {
v3EnvironmentPath,
v3EnvironmentVariablesPath,
v3LogsPath,
v3ErrorsPath,
v3ProjectAlertsPath,
v3ProjectPath,
v3ProjectSettingsGeneralPath,
@@ -112,6 +114,7 @@ import { SideMenuHeader } from "./SideMenuHeader";
import { SideMenuItem } from "./SideMenuItem";
import { SideMenuSection } from "./SideMenuSection";
import { type SideMenuSectionId } from "./sideMenuTypes";
import { IconBugFilled } from "@tabler/icons-react";
/** Get the collapsed state for a specific side menu section from user preferences */
function getSectionCollapsed(
@@ -474,6 +477,17 @@ export function SideMenu({
isCollapsed={isCollapsed}
/>
)}
{(user.admin || user.isImpersonating) && (
<SideMenuItem
name="Errors"
icon={IconBugFilled}
activeIconColor="text-amber-500"
inactiveIconColor="text-amber-500"
to={v3ErrorsPath(organization, project, environment)}
data-action="errors"
isCollapsed={isCollapsed}
/>
)}
<SideMenuItem
name="Query"
icon={TableCellsIcon}
@@ -1,7 +1,8 @@
import { GlobeAltIcon, GlobeAmericasIcon } from "@heroicons/react/20/solid";
import { useRouteLoaderData } from "@remix-run/react";
import { formatDistanceToNow } from "date-fns";
import { Laptop } from "lucide-react";
import { memo, type ReactNode, useMemo, useSyncExternalStore } from "react";
import { memo, type ReactNode, useEffect, useMemo, useState, useSyncExternalStore } from "react";
import { CopyButton } from "./CopyButton";
import { useLocales } from "./LocaleProvider";
import { Paragraph } from "./Paragraph";
@@ -357,6 +358,54 @@ function formatDateTimeAccurate(
return `${datePart} ${timePart}`;
}
type RelativeDateTimeProps = {
date: Date | string;
timeZone?: string;
};
function getRelativeText(date: Date): string {
const text = formatDistanceToNow(date, { addSuffix: true });
return text.charAt(0).toUpperCase() + text.slice(1);
}
export const RelativeDateTime = ({ date, timeZone }: RelativeDateTimeProps) => {
const locales = useLocales();
const userTimeZone = useUserTimeZone();
const realDate = useMemo(() => (typeof date === "string" ? new Date(date) : date), [date]);
const [relativeText, setRelativeText] = useState(() => getRelativeText(realDate));
// Every 60s refresh
useEffect(() => {
const interval = setInterval(() => {
setRelativeText(getRelativeText(realDate));
}, 60_000);
return () => clearInterval(interval);
}, [realDate]);
// On first render
useEffect(() => {
setRelativeText(getRelativeText(realDate));
}, [realDate]);
return (
<SimpleTooltip
button={<span suppressHydrationWarning>{relativeText}</span>}
content={
<TooltipContent
realDate={realDate}
timeZone={timeZone}
localTimeZone={userTimeZone}
locales={locales}
/>
}
side="right"
asChild={true}
/>
);
};
export const DateTimeShort = ({ date, hour12 = true }: DateTimeProps) => {
const locales = useLocales();
const userTimeZone = useUserTimeZone();
@@ -9,7 +9,7 @@ import {
XMarkIcon,
} from "@heroicons/react/20/solid";
import { Form, useFetcher } from "@remix-run/react";
import { IconRotateClockwise2, IconToggleLeft } from "@tabler/icons-react";
import { IconBugFilled, IconRotateClockwise2, IconToggleLeft } from "@tabler/icons-react";
import { MachinePresetName } from "@trigger.dev/core/v3";
import type { BulkActionType, TaskRunStatus, TaskTriggerSource } from "@trigger.dev/database";
import { ListFilterIcon } from "lucide-react";
@@ -181,6 +181,7 @@ export const TaskRunListSearchFilters = z.object({
machines: MachinePresetOrMachinePresetArray.describe(
`Machine presets to filter by (${machines.join(", ")})`
),
errorId: z.string().optional().describe("Error ID to filter runs by (e.g. error_abc123)"),
});
export type TaskRunListSearchFilters = z.infer<typeof TaskRunListSearchFilters>;
@@ -220,6 +221,8 @@ export function filterTitle(filterKey: string) {
return "Machine";
case "versions":
return "Version";
case "errorId":
return "Error ID";
default:
return filterKey;
}
@@ -258,6 +261,8 @@ export function filterIcon(filterKey: string): ReactNode | undefined {
return <MachineDefaultIcon className="size-4" />;
case "versions":
return <IconRotateClockwise2 className="size-4" />;
case "errorId":
return <IconBugFilled className="size-4" />;
default:
return undefined;
}
@@ -304,6 +309,7 @@ export function getRunFiltersFromSearchParams(
searchParams.getAll("versions").filter((v) => v.length > 0).length > 0
? searchParams.getAll("versions")
: undefined,
errorId: searchParams.get("errorId") ?? undefined,
};
const parsed = TaskRunListSearchFilters.safeParse(params);
@@ -344,7 +350,8 @@ export function RunsFilters(props: RunFiltersProps) {
searchParams.has("scheduleId") ||
searchParams.has("queues") ||
searchParams.has("machines") ||
searchParams.has("versions");
searchParams.has("versions") ||
searchParams.has("errorId");
return (
<div className="flex flex-row flex-wrap items-center gap-1">
@@ -380,6 +387,7 @@ const filterTypes = [
{ name: "batch", title: "Batch ID", icon: <Squares2X2Icon className="size-4" /> },
{ name: "schedule", title: "Schedule ID", icon: <ClockIcon className="size-4" /> },
{ name: "bulk", title: "Bulk action", icon: <ListCheckedIcon className="size-4" /> },
{ name: "error", title: "Error ID", icon: <IconBugFilled className="size-4" /> },
] as const;
type FilterType = (typeof filterTypes)[number]["name"];
@@ -434,6 +442,7 @@ function AppliedFilters({ possibleTasks, bulkActions }: RunFiltersProps) {
<AppliedBatchIdFilter />
<AppliedScheduleIdFilter />
<AppliedBulkActionsFilter bulkActions={bulkActions} />
<AppliedErrorIdFilter />
</>
);
}
@@ -470,6 +479,8 @@ function Menu(props: MenuProps) {
return <ScheduleIdDropdown onClose={() => props.setFilterType(undefined)} {...props} />;
case "versions":
return <VersionsDropdown onClose={() => props.setFilterType(undefined)} {...props} />;
case "error":
return <ErrorIdDropdown onClose={() => props.setFilterType(undefined)} {...props} />;
}
}
@@ -655,7 +666,7 @@ function TasksDropdown({
<TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />
}
>
<MiddleTruncate text={item.slug}/>
<MiddleTruncate text={item.slug} />
</SelectItem>
))}
</SelectList>
@@ -1740,3 +1751,121 @@ function AppliedScheduleIdFilter() {
</FilterMenuProvider>
);
}
function ErrorIdDropdown({
trigger,
clearSearchValue,
searchValue,
onClose,
}: {
trigger: ReactNode;
clearSearchValue: () => void;
searchValue: string;
onClose?: () => void;
}) {
const [open, setOpen] = useState<boolean | undefined>();
const { value, replace } = useSearchParams();
const errorIdValue = value("errorId");
const [errorId, setErrorId] = useState(errorIdValue);
const apply = useCallback(() => {
clearSearchValue();
replace({
cursor: undefined,
direction: undefined,
errorId: errorId === "" ? undefined : errorId?.toString(),
});
setOpen(false);
}, [errorId, replace]);
let error: string | undefined = undefined;
if (errorId) {
if (!errorId.startsWith("error_")) {
error = "Error IDs start with 'error_'";
}
}
return (
<SelectProvider virtualFocus={true} open={open} setOpen={setOpen}>
{trigger}
<SelectPopover
hideOnEnter={false}
hideOnEscape={() => {
if (onClose) {
onClose();
return false;
}
return true;
}}
className="max-w-[min(32ch,var(--popover-available-width))]"
>
<div className="flex flex-col gap-4 p-3">
<div className="flex flex-col gap-1">
<Label>Error ID</Label>
<Input
placeholder="error_"
value={errorId ?? ""}
onChange={(e) => setErrorId(e.target.value)}
variant="small"
className="w-[29ch] font-mono"
spellCheck={false}
/>
{error ? <FormError>{error}</FormError> : null}
</div>
<div className="flex justify-between gap-1 border-t border-grid-dimmed pt-3">
<Button variant="tertiary/small" onClick={() => setOpen(false)}>
Cancel
</Button>
<Button
disabled={error !== undefined || !errorId}
variant="secondary/small"
shortcut={{
modifiers: ["mod"],
key: "Enter",
enabledOnInputElements: true,
}}
onClick={() => apply()}
>
Apply
</Button>
</div>
</div>
</SelectPopover>
</SelectProvider>
);
}
function AppliedErrorIdFilter() {
const { value, del } = useSearchParams();
if (value("errorId") === undefined) {
return null;
}
const errorId = value("errorId");
return (
<FilterMenuProvider>
{(search, setSearch) => (
<ErrorIdDropdown
trigger={
<Ariakit.Select render={<div className="group cursor-pointer focus-custom" />}>
<AppliedFilter
label="Error ID"
icon={filterIcon("errorId")}
value={errorId}
onRemove={() => del(["errorId", "cursor", "direction"])}
variant="secondary/small"
/>
</Ariakit.Select>
}
searchValue={search}
clearSearchValue={() => setSearch("")}
/>
)}
</FilterMenuProvider>
);
}
@@ -69,6 +69,7 @@ type RunsTableProps = {
allowSelection?: boolean;
variant?: TableVariant;
disableAdjacentRows?: boolean;
additionalTableState?: Record<string, string>;
};
export function TaskRunsTable({
@@ -81,6 +82,7 @@ export function TaskRunsTable({
isLoading = false,
allowSelection = false,
variant = "dimmed",
additionalTableState,
}: RunsTableProps) {
const organization = useOrganization();
const project = useProject();
@@ -89,8 +91,16 @@ export function TaskRunsTable({
const { isManagedCloud } = useFeatures();
const { value } = useSearchParams();
const location = useOptimisticLocation();
const rootOnly = value("rootOnly") ? `` : `rootOnly=${rootOnlyDefault}`;
const search = rootOnly ? `${rootOnly}&${location.search}` : location.search;
const params = new URLSearchParams(location.search || "");
if (!value("rootOnly")) {
params.set("rootOnly", String(rootOnlyDefault));
}
if (additionalTableState) {
for (const [key, val] of Object.entries(additionalTableState)) {
params.set(key, val);
}
}
const search = params.toString();
/** TableState has to be encoded as a separate URI component, so it's merged under one, 'tableState' param */
const tableStateParam = disableAdjacentRows ? '' : encodeURIComponent(search);
+1
View File
@@ -1199,6 +1199,7 @@ const EnvironmentSchema = z
RUN_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
RUN_REPLICATION_INSERT_STRATEGY: z.enum(["insert", "insert_async"]).default("insert"),
RUN_REPLICATION_DISABLE_PAYLOAD_INSERT: z.string().default("0"),
RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING: z.string().default("0"),
// Clickhouse
CLICKHOUSE_URL: z.string(),
@@ -35,6 +35,7 @@ export async function getRunFiltersFromRequest(request: Request): Promise<Filter
scheduleId,
queues,
machines,
errorId,
} = TaskRunListSearchFilters.parse(s);
return {
@@ -54,5 +55,6 @@ export async function getRunFiltersFromRequest(request: Request): Promise<Filter
cursor: cursor,
queues,
machines,
errorId,
};
}
@@ -0,0 +1,305 @@
import { z } from "zod";
import { type ClickHouse, msToClickHouseInterval } from "@internal/clickhouse";
import { TimeGranularity } from "~/utils/timeGranularity";
import { ErrorId } from "@trigger.dev/core/v3/isomorphic";
import { type PrismaClientOrTransaction } from "@trigger.dev/database";
import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters";
import { type Direction, DirectionSchema } from "~/components/ListPagination";
import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { BasePresenter } from "~/presenters/v3/basePresenter.server";
import {
NextRunListPresenter,
type NextRunList,
} from "~/presenters/v3/NextRunListPresenter.server";
import { sortVersionsDescending } from "~/utils/semver";
const errorGroupGranularity = new TimeGranularity([
{ max: "1h", granularity: "1m" },
{ max: "1d", granularity: "20m" },
{ max: "1w", granularity: "2h" },
{ max: "31d", granularity: "12h" },
{ max: "60d", granularity: "1w" },
{ max: "Infinity", granularity: "30d" },
]);
export type ErrorGroupOptions = {
userId?: string;
projectId: string;
fingerprint: string;
runsPageSize?: number;
period?: string;
from?: number;
to?: number;
cursor?: string;
direction?: Direction;
};
export const ErrorGroupOptionsSchema = z.object({
userId: z.string().optional(),
projectId: z.string(),
fingerprint: z.string(),
runsPageSize: z.number().int().positive().max(1000).optional(),
period: z.string().optional(),
from: z.number().int().nonnegative().optional(),
to: z.number().int().nonnegative().optional(),
cursor: z.string().optional(),
direction: DirectionSchema.optional(),
});
const DEFAULT_RUNS_PAGE_SIZE = 25;
export type ErrorGroupDetail = Awaited<ReturnType<ErrorGroupPresenter["call"]>>;
function parseClickHouseDateTime(value: string): Date {
const asNum = Number(value);
if (!isNaN(asNum) && asNum > 1e12) {
return new Date(asNum);
}
return new Date(value.replace(" ", "T") + "Z");
}
export type ErrorGroupSummary = {
fingerprint: string;
errorType: string;
errorMessage: string;
taskIdentifier: string;
count: number;
firstSeen: Date;
lastSeen: Date;
affectedVersions: string[];
};
export type ErrorGroupOccurrences = Awaited<ReturnType<ErrorGroupPresenter["getOccurrences"]>>;
export type ErrorGroupActivity = ErrorGroupOccurrences["data"];
export class ErrorGroupPresenter extends BasePresenter {
constructor(
private readonly replica: PrismaClientOrTransaction,
private readonly logsClickhouse: ClickHouse,
private readonly clickhouse: ClickHouse
) {
super(undefined, replica);
}
public async call(
organizationId: string,
environmentId: string,
{
userId,
projectId,
fingerprint,
runsPageSize = DEFAULT_RUNS_PAGE_SIZE,
period,
from,
to,
cursor,
direction,
}: ErrorGroupOptions
) {
const displayableEnvironment = await findDisplayableEnvironment(environmentId, userId);
if (!displayableEnvironment) {
throw new ServiceValidationError("No environment found");
}
const time = timeFilterFromTo({
period,
from,
to,
defaultPeriod: "7d",
});
const [summary, affectedVersions, runList] = await Promise.all([
this.getSummary(organizationId, projectId, environmentId, fingerprint),
this.getAffectedVersions(organizationId, projectId, environmentId, fingerprint),
this.getRunList(organizationId, environmentId, {
userId,
projectId,
fingerprint,
pageSize: runsPageSize,
from: time.from.getTime(),
to: time.to.getTime(),
cursor,
direction,
}),
]);
if (summary) {
summary.affectedVersions = affectedVersions;
}
return {
errorGroup: summary,
runList,
filters: {
from: time.from,
to: time.to,
},
};
}
/**
* Returns bucketed occurrence counts for a single fingerprint over a time range.
* Granularity is determined automatically from the range span.
*/
public async getOccurrences(
organizationId: string,
projectId: string,
environmentId: string,
fingerprint: string,
from: Date,
to: Date
): Promise<{
data: Array<{ date: Date; count: number }>;
}> {
const granularityMs = errorGroupGranularity.getTimeGranularityMs(from, to);
const intervalExpr = msToClickHouseInterval(granularityMs);
const queryBuilder = this.logsClickhouse.errors.createOccurrencesQueryBuilder(intervalExpr);
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("error_fingerprint = {fingerprint: String}", { fingerprint });
queryBuilder.where("minute >= toStartOfMinute(fromUnixTimestamp64Milli({fromTimeMs: Int64}))", {
fromTimeMs: from.getTime(),
});
queryBuilder.where("minute <= toStartOfMinute(fromUnixTimestamp64Milli({toTimeMs: Int64}))", {
toTimeMs: to.getTime(),
});
queryBuilder.groupBy("error_fingerprint, bucket_epoch");
queryBuilder.orderBy("bucket_epoch ASC");
const [queryError, records] = await queryBuilder.execute();
if (queryError) {
throw queryError;
}
// Build time buckets covering the full range
const buckets: number[] = [];
const startEpoch = Math.floor(from.getTime() / granularityMs) * (granularityMs / 1000);
const endEpoch = Math.ceil(to.getTime() / 1000);
for (let epoch = startEpoch; epoch <= endEpoch; epoch += granularityMs / 1000) {
buckets.push(epoch);
}
const byBucket = new Map<number, number>();
for (const row of records ?? []) {
byBucket.set(row.bucket_epoch, (byBucket.get(row.bucket_epoch) ?? 0) + row.count);
}
return {
data: buckets.map((epoch) => ({
date: new Date(epoch * 1000),
count: byBucket.get(epoch) ?? 0,
})),
};
}
private async getSummary(
organizationId: string,
projectId: string,
environmentId: string,
fingerprint: string
): Promise<ErrorGroupSummary | undefined> {
const queryBuilder = this.logsClickhouse.errors.listQueryBuilder();
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("error_fingerprint = {fingerprint: String}", { fingerprint });
queryBuilder.groupBy("error_fingerprint, task_identifier");
queryBuilder.limit(1);
const [queryError, records] = await queryBuilder.execute();
if (queryError) {
throw queryError;
}
if (!records || records.length === 0) {
return undefined;
}
const record = records[0];
return {
fingerprint: record.error_fingerprint,
errorType: record.error_type,
errorMessage: record.error_message,
taskIdentifier: record.task_identifier,
count: record.occurrence_count,
firstSeen: parseClickHouseDateTime(record.first_seen),
lastSeen: parseClickHouseDateTime(record.last_seen),
affectedVersions: [],
};
}
/**
* Returns the most recent distinct task_version values for an error fingerprint,
* sorted by semantic version descending (newest first).
* Queries error_occurrences_v1 where task_version is part of the ORDER BY key.
*/
private async getAffectedVersions(
organizationId: string,
projectId: string,
environmentId: string,
fingerprint: string
): Promise<string[]> {
const queryBuilder = this.logsClickhouse.errors.affectedVersionsQueryBuilder();
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("error_fingerprint = {fingerprint: String}", { fingerprint });
queryBuilder.where("task_version != ''");
queryBuilder.limit(100);
const [queryError, records] = await queryBuilder.execute();
if (queryError || !records) {
return [];
}
const versions = records.map((r) => r.task_version).filter((v) => v.length > 0);
return sortVersionsDescending(versions).slice(0, 5);
}
private async getRunList(
organizationId: string,
environmentId: string,
options: {
userId?: string;
projectId: string;
fingerprint: string;
pageSize: number;
from?: number;
to?: number;
cursor?: string;
direction?: Direction;
}
): Promise<NextRunList | undefined> {
const runListPresenter = new NextRunListPresenter(this.replica, this.clickhouse);
const result = await runListPresenter.call(organizationId, environmentId, {
userId: options.userId,
projectId: options.projectId,
rootOnly: false,
errorId: ErrorId.toFriendlyId(options.fingerprint),
pageSize: options.pageSize,
from: options.from,
to: options.to,
cursor: options.cursor,
direction: options.direction,
});
if (result.runs.length === 0) {
return undefined;
}
return result;
}
}
@@ -0,0 +1,408 @@
import { z } from "zod";
import { type ClickHouse, msToClickHouseInterval } from "@internal/clickhouse";
import { TimeGranularity } from "~/utils/timeGranularity";
const errorsListGranularity = new TimeGranularity([
{ max: "2h", granularity: "1m" },
{ max: "2d", granularity: "1h" },
{ max: "2w", granularity: "1d" },
{ max: "3 months", granularity: "1w" },
{ max: "Infinity", granularity: "30d" },
]);
import { type PrismaClientOrTransaction } from "@trigger.dev/database";
import { type Direction } from "~/components/ListPagination";
import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters";
import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
import { getAllTaskIdentifiers } from "~/models/task.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { BasePresenter } from "~/presenters/v3/basePresenter.server";
export type ErrorsListOptions = {
userId?: string;
projectId: string;
// filters
tasks?: string[];
period?: string;
from?: number;
to?: number;
defaultPeriod?: string;
retentionLimitDays?: number;
// search
search?: string;
// pagination
direction?: Direction;
cursor?: string;
pageSize?: number;
};
export const ErrorsListOptionsSchema = z.object({
userId: z.string().optional(),
projectId: z.string(),
tasks: z.array(z.string()).optional(),
period: z.string().optional(),
from: z.number().int().nonnegative().optional(),
to: z.number().int().nonnegative().optional(),
defaultPeriod: z.string().optional(),
retentionLimitDays: z.number().int().positive().optional(),
search: z.string().max(1000).optional(),
direction: z.enum(["forward", "backward"]).optional(),
cursor: z.string().optional(),
pageSize: z.number().int().positive().max(1000).optional(),
});
const DEFAULT_PAGE_SIZE = 25;
export type ErrorsList = Awaited<ReturnType<ErrorsListPresenter["call"]>>;
export type ErrorGroup = ErrorsList["errorGroups"][0];
export type ErrorsListAppliedFilters = ErrorsList["filters"];
export type ErrorOccurrences = Awaited<ReturnType<ErrorsListPresenter["getOccurrences"]>>;
export type ErrorOccurrenceActivity = ErrorOccurrences["data"][string];
type ErrorGroupCursor = {
occurrenceCount: number;
fingerprint: string;
taskIdentifier: string;
};
const ErrorGroupCursorSchema = z.object({
occurrenceCount: z.number(),
fingerprint: z.string(),
taskIdentifier: z.string(),
});
function encodeCursor(cursor: ErrorGroupCursor): string {
return Buffer.from(JSON.stringify(cursor)).toString("base64");
}
function decodeCursor(cursor: string): ErrorGroupCursor | null {
try {
const decoded = Buffer.from(cursor, "base64").toString("utf-8");
const parsed = JSON.parse(decoded);
const validated = ErrorGroupCursorSchema.safeParse(parsed);
if (!validated.success) {
return null;
}
return validated.data as ErrorGroupCursor;
} catch {
return null;
}
}
function cursorFromRow(row: { occurrence_count: number; error_fingerprint: string; task_identifier: string }): string {
return encodeCursor({
occurrenceCount: row.occurrence_count,
fingerprint: row.error_fingerprint,
taskIdentifier: row.task_identifier,
});
}
function parseClickHouseDateTime(value: string): Date {
const asNum = Number(value);
if (!isNaN(asNum) && asNum > 1e12) {
return new Date(asNum);
}
return new Date(value.replace(" ", "T") + "Z");
}
function escapeClickHouseString(val: string): string {
return val.replace(/\\/g, "\\\\").replace(/\//g, "\\/").replace(/%/g, "\\%").replace(/_/g, "\\_");
}
export class ErrorsListPresenter extends BasePresenter {
constructor(
private readonly replica: PrismaClientOrTransaction,
private readonly clickhouse: ClickHouse
) {
super(undefined, replica);
}
public async call(
organizationId: string,
environmentId: string,
{
userId,
projectId,
tasks,
period,
search,
from,
to,
direction,
cursor,
pageSize = DEFAULT_PAGE_SIZE,
defaultPeriod,
retentionLimitDays,
}: ErrorsListOptions
) {
const time = timeFilterFromTo({
period,
from,
to,
defaultPeriod: defaultPeriod ?? "1d",
});
let effectiveFrom = time.from;
let effectiveTo = time.to;
let wasClampedByRetention = false;
if (retentionLimitDays !== undefined && effectiveFrom) {
const retentionCutoffDate = new Date(Date.now() - retentionLimitDays * 24 * 60 * 60 * 1000);
if (effectiveFrom < retentionCutoffDate) {
effectiveFrom = retentionCutoffDate;
wasClampedByRetention = true;
}
}
const hasFilters =
(tasks !== undefined && tasks.length > 0) ||
(search !== undefined && search !== "") ||
!time.isDefault;
const possibleTasksAsync = getAllTaskIdentifiers(this.replica, environmentId);
const [possibleTasks, displayableEnvironment] = await Promise.all([
possibleTasksAsync,
findDisplayableEnvironment(environmentId, userId),
]);
if (!displayableEnvironment) {
throw new ServiceValidationError("No environment found");
}
// Query the per-minute error_occurrences_v1 table for time-scoped counts
const queryBuilder = this.clickhouse.errors.occurrencesListQueryBuilder();
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
// Precise time range filtering via WHERE on the minute column
queryBuilder.where("minute >= toStartOfMinute(fromUnixTimestamp64Milli({fromTimeMs: Int64}))", {
fromTimeMs: effectiveFrom.getTime(),
});
queryBuilder.where("minute <= toStartOfMinute(fromUnixTimestamp64Milli({toTimeMs: Int64}))", {
toTimeMs: effectiveTo.getTime(),
});
if (tasks && tasks.length > 0) {
queryBuilder.where("task_identifier IN {tasks: Array(String)}", { tasks });
}
queryBuilder.groupBy("error_fingerprint, task_identifier");
// Text search via HAVING (operates on aggregated values)
if (search && search.trim() !== "") {
const searchTerm = escapeClickHouseString(search.trim()).toLowerCase();
queryBuilder.having(
"(lower(error_type) like {searchPattern: String} OR lower(error_message) like {searchPattern: String})",
{
searchPattern: `%${searchTerm}%`,
}
);
}
const isBackward = direction === "backward";
const decodedCursor = cursor ? decodeCursor(cursor) : null;
if (decodedCursor) {
const cmp = isBackward ? ">" : "<";
queryBuilder.having(
`(occurrence_count ${cmp} {cursorOccurrenceCount: UInt64}
OR (occurrence_count = {cursorOccurrenceCount: UInt64} AND error_fingerprint ${cmp} {cursorFingerprint: String})
OR (occurrence_count = {cursorOccurrenceCount: UInt64} AND error_fingerprint = {cursorFingerprint: String} AND task_identifier ${cmp} {cursorTaskIdentifier: String}))`,
{
cursorOccurrenceCount: decodedCursor.occurrenceCount,
cursorFingerprint: decodedCursor.fingerprint,
cursorTaskIdentifier: decodedCursor.taskIdentifier,
}
);
}
const sortDir = isBackward ? "ASC" : "DESC";
queryBuilder.orderBy(
`occurrence_count ${sortDir}, error_fingerprint ${sortDir}, task_identifier ${sortDir}`
);
queryBuilder.limit(pageSize + 1);
const [queryError, records] = await queryBuilder.execute();
if (queryError) {
throw queryError;
}
const results = records || [];
const hasMore = results.length > pageSize;
const page = results.slice(0, pageSize);
if (isBackward) {
page.reverse();
}
let nextCursor: string | undefined;
let previousCursor: string | undefined;
if (isBackward) {
previousCursor = hasMore && page.length > 0 ? cursorFromRow(page[0]) : undefined;
nextCursor = page.length > 0 ? cursorFromRow(page[page.length - 1]) : undefined;
} else {
previousCursor = decodedCursor && page.length > 0 ? cursorFromRow(page[0]) : undefined;
nextCursor = hasMore && page.length > 0 ? cursorFromRow(page[page.length - 1]) : undefined;
}
const errorGroups = page;
// Fetch global first_seen / last_seen from the errors_v1 summary table
const fingerprints = errorGroups.map((e) => e.error_fingerprint);
const globalSummaryMap = await this.getGlobalSummary(
organizationId,
projectId,
environmentId,
fingerprints
);
const transformedErrorGroups = errorGroups.map((error) => {
const global = globalSummaryMap.get(error.error_fingerprint);
return {
errorType: error.error_type,
errorMessage: error.error_message,
fingerprint: error.error_fingerprint,
taskIdentifier: error.task_identifier,
firstSeen: global?.firstSeen ?? new Date(),
lastSeen: global?.lastSeen ?? new Date(),
count: error.occurrence_count,
};
});
return {
errorGroups: transformedErrorGroups,
pagination: {
next: nextCursor,
previous: previousCursor,
},
filters: {
tasks,
search,
period: time,
from: effectiveFrom,
to: effectiveTo,
hasFilters,
possibleTasks,
wasClampedByRetention,
},
};
}
/**
* Returns bucketed occurrence counts for the given fingerprints over a time range.
* Granularity is determined automatically from the range span.
*/
public async getOccurrences(
organizationId: string,
projectId: string,
environmentId: string,
fingerprints: string[],
from: Date,
to: Date
): Promise<{
data: Record<string, Array<{ date: Date; count: number }>>;
}> {
if (fingerprints.length === 0) {
return { data: {} };
}
const granularityMs = errorsListGranularity.getTimeGranularityMs(from, to);
const intervalExpr = msToClickHouseInterval(granularityMs);
const queryBuilder = this.clickhouse.errors.createOccurrencesQueryBuilder(intervalExpr);
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("error_fingerprint IN {fingerprints: Array(String)}", { fingerprints });
queryBuilder.where("minute >= toStartOfMinute(fromUnixTimestamp64Milli({fromTimeMs: Int64}))", {
fromTimeMs: from.getTime(),
});
queryBuilder.where("minute <= toStartOfMinute(fromUnixTimestamp64Milli({toTimeMs: Int64}))", {
toTimeMs: to.getTime(),
});
queryBuilder.groupBy("error_fingerprint, bucket_epoch");
queryBuilder.orderBy("error_fingerprint ASC, bucket_epoch ASC");
const [queryError, records] = await queryBuilder.execute();
if (queryError) {
throw queryError;
}
// Build time buckets covering the full range
const buckets: number[] = [];
const startEpoch = Math.floor(from.getTime() / granularityMs) * (granularityMs / 1000);
const endEpoch = Math.ceil(to.getTime() / 1000);
for (let epoch = startEpoch; epoch <= endEpoch; epoch += granularityMs / 1000) {
buckets.push(epoch);
}
// Index results by fingerprint -> epoch -> count
const grouped = new Map<string, Map<number, number>>();
for (const row of records ?? []) {
let byBucket = grouped.get(row.error_fingerprint);
if (!byBucket) {
byBucket = new Map();
grouped.set(row.error_fingerprint, byBucket);
}
byBucket.set(row.bucket_epoch, (byBucket.get(row.bucket_epoch) ?? 0) + row.count);
}
const data: Record<string, Array<{ date: Date; count: number }>> = {};
for (const fp of fingerprints) {
const byBucket = grouped.get(fp);
data[fp] = buckets.map((epoch) => ({
date: new Date(epoch * 1000),
count: byBucket?.get(epoch) ?? 0,
}));
}
return { data };
}
/**
* Fetches global first_seen / last_seen for a set of fingerprints from errors_v1.
*/
private async getGlobalSummary(
organizationId: string,
projectId: string,
environmentId: string,
fingerprints: string[]
): Promise<Map<string, { firstSeen: Date; lastSeen: Date }>> {
const result = new Map<string, { firstSeen: Date; lastSeen: Date }>();
if (fingerprints.length === 0) return result;
const queryBuilder = this.clickhouse.errors.listQueryBuilder();
queryBuilder.where("organization_id = {organizationId: String}", { organizationId });
queryBuilder.where("project_id = {projectId: String}", { projectId });
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
queryBuilder.where("error_fingerprint IN {fingerprints: Array(String)}", { fingerprints });
queryBuilder.groupBy("error_fingerprint, task_identifier");
const [queryError, records] = await queryBuilder.execute();
if (queryError || !records) return result;
for (const record of records) {
const firstSeen = parseClickHouseDateTime(record.first_seen);
const lastSeen = parseClickHouseDateTime(record.last_seen);
const existing = result.get(record.error_fingerprint);
if (existing) {
if (firstSeen < existing.firstSeen) existing.firstSeen = firstSeen;
if (lastSeen > existing.lastSeen) existing.lastSeen = lastSeen;
} else {
result.set(record.error_fingerprint, { firstSeen, lastSeen });
}
}
return result;
}
}
@@ -33,6 +33,7 @@ export type RunListOptions = {
runId?: string[];
queues?: string[];
machines?: MachinePresetName[];
errorId?: string;
//pagination
direction?: Direction;
cursor?: string;
@@ -70,6 +71,7 @@ export class NextRunListPresenter {
runId,
queues,
machines,
errorId,
from,
to,
direction = "forward",
@@ -97,6 +99,7 @@ export class NextRunListPresenter {
(runId !== undefined && runId.length > 0) ||
(queues !== undefined && queues.length > 0) ||
(machines !== undefined && machines.length > 0) ||
(errorId !== undefined && errorId !== "") ||
typeof isTest === "boolean" ||
rootOnly === true ||
!time.isDefault;
@@ -182,6 +185,7 @@ export class NextRunListPresenter {
bulkId,
queues,
machines,
errorId,
page: {
size: pageSize,
cursor,
@@ -0,0 +1,486 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { type MetaFunction } from "@remix-run/react";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { TypedAwait, typeddefer, useTypedLoaderData } from "remix-typedjson";
import { requireUser } from "~/services/session.server";
import {
EnvironmentParamSchema,
v3CreateBulkActionPath,
v3ErrorsPath,
v3RunsPath,
} from "~/utils/pathBuilder";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import {
ErrorGroupPresenter,
type ErrorGroupActivity,
type ErrorGroupOccurrences,
type ErrorGroupSummary,
} from "~/presenters/v3/ErrorGroupPresenter.server";
import { type NextRunList } from "~/presenters/v3/NextRunListPresenter.server";
import { $replica } from "~/db.server";
import { logsClickhouseClient, clickhouseClient } from "~/services/clickhouseInstance.server";
import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
import { PageBody } from "~/components/layout/AppLayout";
import { Suspense, useMemo } from "react";
import { Spinner } from "~/components/primitives/Spinner";
import { Paragraph } from "~/components/primitives/Paragraph";
import { Callout } from "~/components/primitives/Callout";
import { Header1, Header2, Header3 } from "~/components/primitives/Headers";
import { formatDistanceToNow } from "date-fns";
import { formatNumberCompact } from "~/utils/numberFormatter";
import * as Property from "~/components/primitives/PropertyTable";
import { TaskRunsTable } from "~/components/runs/v3/TaskRunsTable";
import { DateTime, RelativeDateTime } from "~/components/primitives/DateTime";
import { ErrorId } from "@trigger.dev/core/v3/isomorphic";
import { Chart, type ChartConfig } from "~/components/primitives/charts/ChartCompound";
import { TimeFilter, timeFilterFromTo } from "~/components/runs/v3/SharedFilters";
import { useOptimisticLocation } from "~/hooks/useOptimisticLocation";
import { DirectionSchema, ListPagination } from "~/components/ListPagination";
import { LinkButton } from "~/components/primitives/Buttons";
import { ListCheckedIcon } from "~/assets/icons/ListCheckedIcon";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { useEnvironment } from "~/hooks/useEnvironment";
import { RunsIcon } from "~/assets/icons/RunsIcon";
import { TaskRunListSearchFilters } from "~/components/runs/v3/RunFilters";
import { useSearchParams } from "~/hooks/useSearchParam";
import { CopyableText } from "~/components/primitives/CopyableText";
export const meta: MetaFunction<typeof loader> = ({ data }) => {
return [
{
title: `Error Details | Trigger.dev`,
},
];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const user = await requireUser(request);
const userId = user.id;
const { projectParam, organizationSlug, envParam } = EnvironmentParamSchema.parse(params);
const fingerprint = params.fingerprint;
if (!fingerprint) {
throw new Response("Fingerprint parameter is required", { status: 400 });
}
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
throw new Response("Project not found", { status: 404 });
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
throw new Response("Environment not found", { status: 404 });
}
const url = new URL(request.url);
const period = url.searchParams.get("period") ?? undefined;
const fromStr = url.searchParams.get("from");
const toStr = url.searchParams.get("to");
const from = fromStr ? parseInt(fromStr, 10) : undefined;
const to = toStr ? parseInt(toStr, 10) : undefined;
const cursor = url.searchParams.get("cursor") ?? undefined;
const directionRaw = url.searchParams.get("direction") ?? undefined;
const direction = directionRaw ? DirectionSchema.parse(directionRaw) : undefined;
const presenter = new ErrorGroupPresenter($replica, logsClickhouseClient, clickhouseClient);
const detailPromise = presenter
.call(project.organizationId, environment.id, {
userId,
projectId: project.id,
fingerprint,
period,
from,
to,
cursor,
direction,
})
.catch((error) => {
if (error instanceof ServiceValidationError) {
return { error: error.message };
}
throw error;
});
const time = timeFilterFromTo({ period, from, to, defaultPeriod: "7d" });
const activityPromise = presenter
.getOccurrences(
project.organizationId,
project.id,
environment.id,
fingerprint,
time.from,
time.to
)
.catch(() => ({ data: [] as ErrorGroupActivity }));
return typeddefer({
data: detailPromise,
activity: activityPromise,
organizationSlug,
projectParam,
envParam,
fingerprint,
});
};
export default function Page() {
const { data, activity, organizationSlug, projectParam, envParam, fingerprint } =
useTypedLoaderData<typeof loader>();
const location = useOptimisticLocation();
const searchParams = new URLSearchParams(location.search);
const errorsPath = useMemo(() => {
const base = v3ErrorsPath(
{ slug: organizationSlug },
{ slug: projectParam },
{ slug: envParam }
);
const carry = new URLSearchParams();
const period = searchParams.get("period");
const from = searchParams.get("from");
const to = searchParams.get("to");
if (period) carry.set("period", period);
if (from) carry.set("from", from);
if (to) carry.set("to", to);
const qs = carry.toString();
return qs ? `${base}?${qs}` : base;
}, [organizationSlug, projectParam, envParam, searchParams.toString()]);
return (
<>
<NavBar>
<PageTitle
backButton={{
to: errorsPath,
text: "Errors",
}}
title={<span className="font-mono text-xs">{ErrorId.toFriendlyId(fingerprint)}</span>}
/>
</NavBar>
<PageBody scrollable={false}>
<Suspense
fallback={
<div className="my-2 flex items-center justify-center">
<div className="mx-auto flex items-center gap-2">
<Spinner />
<Paragraph variant="small">Loading error details</Paragraph>
</div>
</div>
}
>
<TypedAwait
resolve={data}
errorElement={
<div className="flex items-center justify-center px-3 py-12">
<Callout variant="error" className="max-w-fit">
Unable to load error details. Please refresh the page or try again in a moment.
</Callout>
</div>
}
>
{(result) => {
if ("error" in result) {
return (
<div className="flex items-center justify-center px-3 py-12">
<Callout variant="error" className="max-w-fit">
{result.error}
</Callout>
</div>
);
}
return (
<ErrorGroupDetail
errorGroup={result.errorGroup}
runList={result.runList}
activity={activity}
organizationSlug={organizationSlug}
projectParam={projectParam}
envParam={envParam}
fingerprint={fingerprint}
/>
);
}}
</TypedAwait>
</Suspense>
</PageBody>
</>
);
}
function ErrorGroupDetail({
errorGroup,
runList,
activity,
organizationSlug,
projectParam,
envParam,
fingerprint,
}: {
errorGroup: ErrorGroupSummary | undefined;
runList: NextRunList | undefined;
activity: Promise<ErrorGroupOccurrences>;
organizationSlug: string;
projectParam: string;
envParam: string;
fingerprint: string;
}) {
const { value } = useSearchParams();
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
if (!errorGroup) {
return (
<div className="flex h-full items-center justify-center">
<div className="text-center">
<Header3 className="mb-2">Error not found</Header3>
<Paragraph variant="small">
This error group does not exist or has no instances.
</Paragraph>
</div>
</div>
);
}
const fromValue = value("from") ?? undefined;
const toValue = value("to") ?? undefined;
const filters: TaskRunListSearchFilters = {
period: value("period") ?? undefined,
from: fromValue ? parseInt(fromValue, 10) : undefined,
to: toValue ? parseInt(toValue, 10) : undefined,
rootOnly: false,
errorId: ErrorId.toFriendlyId(fingerprint),
};
return (
<div className="grid h-full grid-rows-[auto_12rem_1fr] overflow-hidden">
{/* Error Summary */}
<div className="flex flex-col gap-2 border-b border-grid-bright bg-background-bright p-4">
<div className="flex flex-col gap-0.5">
<Header2>{errorGroup.errorMessage}</Header2>
<Header3>{formatNumberCompact(errorGroup.count)} total occurrences</Header3>
</div>
<div className="grid grid-cols-[auto_auto_auto_1fr] gap-x-12 gap-y-0.5">
<Property.Table>
<Property.Item>
<Property.Label>ID</Property.Label>
<Property.Value>
<CopyableText value={ErrorId.toFriendlyId(errorGroup.fingerprint)} />
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Task</Property.Label>
<Property.Value>
<CopyableText value={errorGroup.taskIdentifier} />
</Property.Value>
</Property.Item>
</Property.Table>
<Property.Table>
<Property.Item>
<Property.Label>First seen</Property.Label>
<Property.Value>
<DateTime date={errorGroup.firstSeen} />
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Last seen</Property.Label>
<Property.Value>
<RelativeDateTime date={errorGroup.lastSeen} />
</Property.Value>
</Property.Item>
</Property.Table>
<Property.Table>
{errorGroup.affectedVersions.length > 0 && (
<Property.Item>
<Property.Label>Affected versions</Property.Label>
<Property.Value>
<span className="font-mono text-xs">
{errorGroup.affectedVersions.join(", ")}
</span>
</Property.Value>
</Property.Item>
)}
</Property.Table>
</div>
</div>
{/* Activity chart */}
<div className="flex flex-col gap-3 overflow-hidden border-b border-grid-bright px-4 py-3">
<div className="flex items-center">
<TimeFilter defaultPeriod="7d" labelName="Occurred" />
</div>
<Suspense fallback={<ActivityChartBlankState />}>
<TypedAwait resolve={activity} errorElement={<ActivityChartBlankState />}>
{(result) =>
result.data.length > 0 ? (
<ActivityChart activity={result.data} />
) : (
<ActivityChartBlankState />
)
}
</TypedAwait>
</Suspense>
</div>
{/* Runs Table */}
<div className="flex flex-col gap-1 overflow-y-hidden">
<div className="flex items-center justify-between px-4">
<Header3 className="mb-1 mt-2">Runs</Header3>
{runList && (
<div className="flex items-center gap-2">
<LinkButton
variant="secondary/small"
to={v3RunsPath(organization, project, environment, filters)}
LeadingIcon={RunsIcon}
>
View all runs
</LinkButton>
<LinkButton
variant="secondary/small"
to={v3CreateBulkActionPath(
organization,
project,
environment,
filters,
"filter",
"replay"
)}
LeadingIcon={ListCheckedIcon}
>
Bulk replay
</LinkButton>
<ListPagination list={runList} />
</div>
)}
</div>
{runList ? (
<TaskRunsTable
total={runList.runs.length}
hasFilters={false}
filters={{
tasks: [],
versions: [],
statuses: [],
from: undefined,
to: undefined,
}}
runs={runList.runs}
isLoading={false}
variant="dimmed"
additionalTableState={{ errorId: ErrorId.toFriendlyId(fingerprint) }}
/>
) : (
<Paragraph variant="small" className="p-4 text-text-dimmed">
No runs found for this error.
</Paragraph>
)}
</div>
</div>
);
}
const activityChartConfig: ChartConfig = {
count: {
label: "Occurrences",
color: "#6366F1",
},
};
function ActivityChart({ activity }: { activity: ErrorGroupActivity }) {
const data = useMemo(
() =>
activity.map((d) => ({
...d,
__timestamp: d.date instanceof Date ? d.date.getTime() : new Date(d.date).getTime(),
})),
[activity]
);
const midnightTicks = useMemo(() => {
const ticks: number[] = [];
for (const d of data) {
const date = new Date(d.__timestamp);
if (date.getHours() === 0 && date.getMinutes() === 0) {
ticks.push(d.__timestamp);
}
}
return ticks;
}, [data]);
const xAxisFormatter = useMemo(() => {
return (value: number) => {
const date = new Date(value);
return date.toLocaleDateString("en-US", {
month: "short",
day: "numeric",
hour: "2-digit",
minute: "2-digit",
hour12: false,
});
};
}, []);
const tooltipLabelFormatter = useMemo(() => {
return (_label: string, payload: Array<{ payload?: Record<string, unknown> }>) => {
const timestamp = payload[0]?.payload?.__timestamp as number | undefined;
if (timestamp) {
const date = new Date(timestamp);
return date.toLocaleString("en-US", {
month: "short",
day: "numeric",
year: "numeric",
hour: "2-digit",
minute: "2-digit",
hour12: false,
});
}
return _label;
};
}, []);
return (
<Chart.Root
config={activityChartConfig}
data={data}
dataKey="__timestamp"
series={["count"]}
fillContainer
>
<Chart.Bar
xAxisProps={{
tickFormatter: xAxisFormatter,
ticks: midnightTicks,
height: 40,
}}
yAxisProps={{
width: 30,
tickMargin: 4,
}}
tooltipLabelFormatter={tooltipLabelFormatter}
/>
</Chart.Root>
);
}
function ActivityChartBlankState() {
return (
<div className="flex min-h-0 flex-1 items-end gap-px rounded-sm">
{[...Array(42)].map((_, i) => (
<div key={i} className="h-full flex-1 bg-charcoal-850" />
))}
</div>
);
}
@@ -0,0 +1,479 @@
import { XMarkIcon } from "@heroicons/react/20/solid";
import { Form, type MetaFunction } from "@remix-run/react";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { ErrorId } from "@trigger.dev/core/v3/isomorphic";
import { Suspense, useMemo } from "react";
import {
Bar,
BarChart,
ReferenceLine,
ResponsiveContainer,
Tooltip,
YAxis,
type TooltipProps,
} from "recharts";
import { TypedAwait, typeddefer, useTypedLoaderData } from "remix-typedjson";
import { PageBody } from "~/components/layout/AppLayout";
import { LogsSearchInput } from "~/components/logs/LogsSearchInput";
import { LogsTaskFilter } from "~/components/logs/LogsTaskFilter";
import { Button } from "~/components/primitives/Buttons";
import { Callout } from "~/components/primitives/Callout";
import { formatDateTime, RelativeDateTime } from "~/components/primitives/DateTime";
import { Header3 } from "~/components/primitives/Headers";
import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
import { Paragraph } from "~/components/primitives/Paragraph";
import { Spinner } from "~/components/primitives/Spinner";
import {
CopyableTableCell,
Table,
TableBody,
TableCell,
TableCellChevron,
TableHeader,
TableHeaderCell,
TableRow,
} from "~/components/primitives/Table";
import TooltipPortal from "~/components/primitives/TooltipPortal";
import { TimeFilter } from "~/components/runs/v3/SharedFilters";
import { $replica } from "~/db.server";
import { useOptimisticLocation } from "~/hooks/useOptimisticLocation";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import {
ErrorsListPresenter,
type ErrorGroup,
type ErrorOccurrenceActivity,
type ErrorOccurrences,
type ErrorsList as ErrorsListData,
} from "~/presenters/v3/ErrorsListPresenter.server";
import { logsClickhouseClient } from "~/services/clickhouseInstance.server";
import { getCurrentPlan } from "~/services/platform.v3.server";
import { requireUser } from "~/services/session.server";
import { ListPagination } from "~/components/ListPagination";
import { formatNumberCompact } from "~/utils/numberFormatter";
import { EnvironmentParamSchema, v3ErrorPath } from "~/utils/pathBuilder";
import { ServiceValidationError } from "~/v3/services/baseService.server";
export const meta: MetaFunction = () => {
return [
{
title: `Errors | Trigger.dev`,
},
];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const user = await requireUser(request);
const userId = user.id;
const { projectParam, organizationSlug, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
throw new Response("Project not found", { status: 404 });
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
throw new Response("Environment not found", { status: 404 });
}
const url = new URL(request.url);
const tasks = url.searchParams.getAll("tasks").filter((t) => t.length > 0);
const search = url.searchParams.get("search") ?? undefined;
const period = url.searchParams.get("period") ?? undefined;
const fromStr = url.searchParams.get("from");
const toStr = url.searchParams.get("to");
const from = fromStr ? parseInt(fromStr, 10) : undefined;
const to = toStr ? parseInt(toStr, 10) : undefined;
const cursor = url.searchParams.get("cursor") ?? undefined;
const directionRaw = url.searchParams.get("direction");
const direction =
directionRaw === "forward" || directionRaw === "backward" ? directionRaw : undefined;
const plan = await getCurrentPlan(project.organizationId);
const retentionLimitDays = plan?.v3Subscription?.plan?.limits.logRetentionDays.number ?? 30;
const presenter = new ErrorsListPresenter($replica, logsClickhouseClient);
const listPromise = presenter
.call(project.organizationId, environment.id, {
userId,
projectId: project.id,
tasks: tasks.length > 0 ? tasks : undefined,
search,
period,
from,
to,
cursor,
direction,
defaultPeriod: "1d",
retentionLimitDays,
})
.catch((error) => {
if (error instanceof ServiceValidationError) {
return { error: error.message };
}
throw error;
});
const occurrencesPromise = listPromise.then((result) => {
if ("error" in result) return { data: {} };
const fingerprints = result.errorGroups.map((g) => g.fingerprint);
if (fingerprints.length === 0) return { data: {} };
return presenter.getOccurrences(
project.organizationId,
project.id,
environment.id,
fingerprints,
result.filters.from,
result.filters.to
);
});
return typeddefer({
data: listPromise,
occurrences: occurrencesPromise,
defaultPeriod: "1d",
retentionLimitDays,
organizationSlug,
projectParam,
envParam,
});
};
export default function Page() {
const {
data,
occurrences,
defaultPeriod,
retentionLimitDays,
organizationSlug,
projectParam,
envParam,
} = useTypedLoaderData<typeof loader>();
return (
<>
<NavBar>
<PageTitle title="Errors" />
</NavBar>
<PageBody scrollable={false}>
<Suspense
fallback={
<div className="grid h-full max-h-full grid-rows-[2.5rem_auto] overflow-hidden">
<div className="border-b border-grid-bright" />
<div className="my-2 flex items-center justify-center">
<div className="mx-auto flex items-center gap-2">
<Spinner />
<Paragraph variant="small">Loading errors</Paragraph>
</div>
</div>
</div>
}
>
<TypedAwait
resolve={data}
errorElement={
<div className="grid h-full max-h-full grid-rows-[2.5rem_auto_1fr] overflow-hidden">
<FiltersBar defaultPeriod={defaultPeriod} retentionLimitDays={retentionLimitDays} />
<div className="flex items-center justify-center px-3 py-12">
<Callout variant="error" className="max-w-fit">
Unable to load errors. Please refresh the page or try again in a moment.
</Callout>
</div>
</div>
}
>
{(result) => {
if ("error" in result) {
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_auto_1fr] overflow-hidden">
<FiltersBar
defaultPeriod={defaultPeriod}
retentionLimitDays={retentionLimitDays}
/>
<div className="flex items-center justify-center px-3 py-12">
<Callout variant="error" className="max-w-fit">
{result.error}
</Callout>
</div>
</div>
);
}
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_1fr] overflow-hidden">
<FiltersBar
list={result}
defaultPeriod={defaultPeriod}
retentionLimitDays={retentionLimitDays}
/>
<ErrorsList
errorGroups={result.errorGroups}
occurrences={occurrences}
organizationSlug={organizationSlug}
projectParam={projectParam}
envParam={envParam}
/>
</div>
);
}}
</TypedAwait>
</Suspense>
</PageBody>
</>
);
}
function FiltersBar({
list,
defaultPeriod,
retentionLimitDays,
}: {
list?: ErrorsListData;
defaultPeriod?: string;
retentionLimitDays: number;
}) {
const location = useOptimisticLocation();
const searchParams = new URLSearchParams(location.search);
const hasFilters =
searchParams.has("tasks") ||
searchParams.has("search") ||
searchParams.has("period") ||
searchParams.has("from") ||
searchParams.has("to");
return (
<div className="flex items-start justify-between gap-x-2 border-b border-grid-bright p-2">
<div className="flex flex-row flex-wrap items-center gap-1">
{list ? (
<>
<LogsTaskFilter possibleTasks={list.filters.possibleTasks} />
<TimeFilter
defaultPeriod={defaultPeriod}
maxPeriodDays={retentionLimitDays}
labelName="Occurred"
/>
<LogsSearchInput placeholder="Search errors..." />
{hasFilters && (
<Form className="h-6">
<Button
variant="secondary/small"
LeadingIcon={XMarkIcon}
tooltip="Clear all filters"
/>
</Form>
)}
</>
) : (
<>
<LogsTaskFilter possibleTasks={[]} />
<TimeFilter defaultPeriod={defaultPeriod} maxPeriodDays={retentionLimitDays} />
<LogsSearchInput placeholder="Search errors..." />
{hasFilters && (
<Form className="h-6">
<Button
variant="secondary/small"
LeadingIcon={XMarkIcon}
tooltip="Clear all filters"
/>
</Form>
)}
</>
)}
</div>
{list && <ListPagination list={list} />}
</div>
);
}
function ErrorsList({
errorGroups,
occurrences,
organizationSlug,
projectParam,
envParam,
}: {
errorGroups: ErrorGroup[];
occurrences: Promise<ErrorOccurrences>;
organizationSlug: string;
projectParam: string;
envParam: string;
}) {
if (errorGroups.length === 0) {
return (
<div className="flex h-full items-center justify-center">
<div className="text-center">
<Header3 className="mb-2">No errors found</Header3>
<Paragraph variant="small">
No errors have been recorded in the selected time period.
</Paragraph>
</div>
</div>
);
}
return (
<Table containerClassName="max-h-full pb-[2.5rem]">
<TableHeader>
<TableRow>
<TableHeaderCell>ID</TableHeaderCell>
<TableHeaderCell>Task</TableHeaderCell>
<TableHeaderCell>Error</TableHeaderCell>
<TableHeaderCell>Occurrences</TableHeaderCell>
<TableHeaderCell>Activity</TableHeaderCell>
<TableHeaderCell>First seen</TableHeaderCell>
<TableHeaderCell>Last seen</TableHeaderCell>
</TableRow>
</TableHeader>
<TableBody>
{errorGroups.map((errorGroup) => (
<ErrorGroupRow
key={errorGroup.fingerprint}
errorGroup={errorGroup}
occurrences={occurrences}
organizationSlug={organizationSlug}
projectParam={projectParam}
envParam={envParam}
/>
))}
</TableBody>
</Table>
);
}
function ErrorGroupRow({
errorGroup,
occurrences,
organizationSlug,
projectParam,
envParam,
}: {
errorGroup: ErrorGroup;
occurrences: Promise<ErrorOccurrences>;
organizationSlug: string;
projectParam: string;
envParam: string;
}) {
const location = useOptimisticLocation();
const searchParams = new URLSearchParams(location.search);
const errorPath = useMemo(() => {
const base = v3ErrorPath(
{ slug: organizationSlug },
{ slug: projectParam },
{ slug: envParam },
{ fingerprint: errorGroup.fingerprint }
);
const carry = new URLSearchParams();
const period = searchParams.get("period");
const from = searchParams.get("from");
const to = searchParams.get("to");
if (period) carry.set("period", period);
if (from) carry.set("from", from);
if (to) carry.set("to", to);
const qs = carry.toString();
return qs ? `${base}?${qs}` : base;
}, [organizationSlug, projectParam, envParam, errorGroup.fingerprint, searchParams.toString()]);
const errorMessage = `${errorGroup.errorMessage}`;
return (
<TableRow>
<CopyableTableCell to={errorPath} value={ErrorId.toFriendlyId(errorGroup.fingerprint)}>
{errorGroup.fingerprint.slice(-8)}
</CopyableTableCell>
<TableCell to={errorPath}>{errorGroup.taskIdentifier}</TableCell>
<CopyableTableCell to={errorPath} className="font-mono" value={errorMessage}>
{errorMessage}
</CopyableTableCell>
<TableCell to={errorPath}>{errorGroup.count.toLocaleString()}</TableCell>
<TableCell to={errorPath} actionClassName="py-1.5">
<Suspense fallback={<ErrorActivityBlankState />}>
<TypedAwait resolve={occurrences} errorElement={<ErrorActivityBlankState />}>
{(result) => {
const activity = result.data[errorGroup.fingerprint];
return activity ? (
<ErrorActivityGraph activity={activity} />
) : (
<ErrorActivityBlankState />
);
}}
</TypedAwait>
</Suspense>
</TableCell>
<TableCell to={errorPath}>
<RelativeDateTime date={errorGroup.firstSeen} />
</TableCell>
<TableCell to={errorPath}>
<RelativeDateTime date={errorGroup.lastSeen} />
</TableCell>
</TableRow>
);
}
function ErrorActivityGraph({ activity }: { activity: ErrorOccurrenceActivity }) {
const maxCount = Math.max(...activity.map((d) => d.count));
return (
<div className="flex items-start gap-1.5">
<div className="h-6 w-[10.25rem] rounded-sm">
<ResponsiveContainer width="100%" height="100%">
<BarChart data={activity} margin={{ top: 0, right: 0, left: 0, bottom: 0 }}>
<YAxis domain={[0, maxCount || 1]} hide />
<Tooltip
cursor={{ fill: "transparent" }}
content={<ErrorActivityTooltip />}
allowEscapeViewBox={{ x: true, y: true }}
wrapperStyle={{ zIndex: 1000 }}
animationDuration={0}
/>
<Bar dataKey="count" fill="#6366F1" strokeWidth={0} isAnimationActive={false} />
<ReferenceLine y={0} stroke="#2C3034" strokeWidth={1} />
{maxCount > 0 && (
<ReferenceLine y={maxCount} stroke="#4D525B" strokeDasharray="4 4" strokeWidth={1} />
)}
</BarChart>
</ResponsiveContainer>
</div>
<span className="-mt-1 text-xxs tabular-nums text-text-dimmed">
{formatNumberCompact(maxCount)}
</span>
</div>
);
}
const ErrorActivityTooltip = ({ active, payload }: TooltipProps<number, string>) => {
if (active && payload && payload.length > 0) {
const entry = payload[0].payload as { date: Date; count: number };
const date = entry.date instanceof Date ? entry.date : new Date(entry.date);
const formattedDate = formatDateTime(date, "UTC", [], false, true);
return (
<TooltipPortal active={active}>
<div className="rounded-sm border border-grid-bright bg-background-dimmed px-3 py-2">
<Header3 className="border-b border-b-charcoal-650 pb-2">{formattedDate}</Header3>
<div className="mt-2 text-xs text-text-bright">
<span className="tabular-nums">{entry.count}</span>{" "}
<span className="text-text-dimmed">
{entry.count === 1 ? "occurrence" : "occurrences"}
</span>
</div>
</div>
</TooltipPortal>
);
}
return null;
};
function ErrorActivityBlankState() {
return (
<div className="flex h-6 w-[5.125rem] items-end gap-px rounded-sm">
{[...Array(24)].map((_, i) => (
<div key={i} className="h-full flex-1 bg-[#212327]" />
))}
</div>
);
}
@@ -0,0 +1,10 @@
import { Outlet } from "@remix-run/react";
import { PageContainer } from "~/components/layout/AppLayout";
export default function Page() {
return (
<PageContainer>
<Outlet />
</PageContainer>
);
}
@@ -68,6 +68,7 @@ function initializeRunsReplicationInstance() {
insertMaxDelayMs: env.RUN_REPLICATION_INSERT_MAX_DELAY_MS,
insertStrategy: env.RUN_REPLICATION_INSERT_STRATEGY,
disablePayloadInsert: env.RUN_REPLICATION_DISABLE_PAYLOAD_INSERT === "1",
disableErrorFingerprinting: env.RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING === "1",
});
if (env.RUN_REPLICATION_ENABLED === "1") {
@@ -27,6 +27,7 @@ import { nanoid } from "nanoid";
import EventEmitter from "node:events";
import pLimit from "p-limit";
import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
import { calculateErrorFingerprint } from "~/utils/errorFingerprinting";
interface TransactionEvent<T = any> {
tag: "insert" | "update" | "delete";
@@ -70,6 +71,7 @@ export type RunsReplicationServiceOptions = {
insertBaseDelayMs?: number;
insertMaxDelayMs?: number;
disablePayloadInsert?: boolean;
disableErrorFingerprinting?: boolean;
};
type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -115,6 +117,7 @@ export class RunsReplicationService {
private _insertMaxDelayMs: number;
private _insertStrategy: "insert" | "insert_async";
private _disablePayloadInsert: boolean;
private _disableErrorFingerprinting: boolean;
// Metrics
private _replicationLagHistogram: Histogram;
@@ -189,6 +192,7 @@ export class RunsReplicationService {
this._insertStrategy = options.insertStrategy ?? "insert";
this._disablePayloadInsert = options.disablePayloadInsert ?? false;
this._disableErrorFingerprinting = options.disableErrorFingerprinting ?? false;
this._replicationClient = new LogicalReplicationClient({
pgConfig: {
@@ -852,6 +856,15 @@ export class RunsReplicationService {
_version: bigint
): Promise<TaskRunInsertArray> {
const output = await this.#prepareJson(run.output, run.outputType);
const errorData = { data: run.error };
// Calculate error fingerprint for failed runs
const errorFingerprint = (
!this._disableErrorFingerprinting &&
['SYSTEM_FAILURE', 'CRASHED', 'INTERRUPTED', 'COMPLETED_WITH_ERRORS', 'TIMED_OUT'].includes(run.status)
)
? calculateErrorFingerprint(run.error)
: '';
// Return array matching TASK_RUN_COLUMNS order
return [
@@ -880,7 +893,8 @@ export class RunsReplicationService {
run.costInCents ?? 0, // cost_in_cents
run.baseCostInCents ?? 0, // base_cost_in_cents
output, // output
{ data: run.error }, // error
errorData, // error
errorFingerprint, // error_fingerprint
run.runTags ?? [], // tags
run.taskVersion ?? "", // task_version
run.sdkVersion ?? "", // sdk_version
@@ -1,5 +1,5 @@
import { type ClickhouseQueryBuilder } from "@internal/clickhouse";
import { RunId } from "@trigger.dev/core/v3/isomorphic";
import { ErrorId, RunId } from "@trigger.dev/core/v3/isomorphic";
import {
type FilterRunsOptions,
type IRunsRepository,
@@ -328,4 +328,10 @@ function applyRunFiltersToQueryBuilder<T>(
machines: options.machines,
});
}
if (options.errorId) {
queryBuilder.where("error_fingerprint = {errorFingerprint: String}", {
errorFingerprint: ErrorId.toId(options.errorId),
});
}
}
@@ -44,6 +44,7 @@ const RunListInputOptionsSchema = z.object({
bulkId: z.string().optional(),
queues: z.array(z.string()).optional(),
machines: MachinePresetName.array().optional(),
errorId: z.string().optional(),
});
export type RunListInputOptions = z.infer<typeof RunListInputOptionsSchema>;
@@ -0,0 +1,93 @@
import { createHash } from "node:crypto";
/**
* Calculate error fingerprint using Sentry-style normalization.
* Groups similar errors together by normalizing dynamic values.
*/
export function calculateErrorFingerprint(error: unknown): string {
if (!error || typeof error !== "object" || Array.isArray(error)) return "";
// This is a but ugly but…
// 1. We can't use a schema here because it's a hot path and needs to be fast.
// 2. It won't be an instanceof Error because it's from the database.
const errorObj = error as any;
const errorType = String(errorObj.type || errorObj.name || "Error");
const message = String(errorObj.message || "");
const stack = String(errorObj.stack || errorObj.stacktrace || "");
// Normalize message to group similar errors
const normalizedMessage = normalizeErrorMessage(message);
// Extract and normalize first few stack frames
const normalizedStack = normalizeStackTrace(stack);
// Create fingerprint from type + normalized message + stack
const fingerprintInput = `${errorType}:${normalizedMessage}:${normalizedStack}`;
// Use SHA-256 hash, take first 16 chars for compact storage
return createHash("sha256").update(fingerprintInput).digest("hex").substring(0, 16);
}
/**
* Normalize error message by replacing dynamic values with placeholders.
* This allows similar errors to be grouped together.
*/
export function normalizeErrorMessage(message: string): string {
if (!message) return "";
return (
message
// UUIDs (8-4-4-4-12 format)
.replace(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi, "<uuid>")
// Run IDs (run_xxxxx format)
.replace(/run_[a-zA-Z0-9]+/g, "<run-id>")
// Task run friendly IDs (task_xxxxx or similar)
.replace(/\b[a-z]+_[a-zA-Z0-9]{8,}\b/g, "<id>")
// --- Specific patterns must run before generic numeric/path replacements ---
// ISO 8601 timestamps
.replace(/\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z?/g, "<timestamp>")
// Unix timestamps (10 or 13 digits)
.replace(/\b\d{10,13}\b/g, "<timestamp>")
// URLs (before path regex, which would strip the URL's path component)
.replace(/https?:\/\/[^\s]+/g, "<url>")
// --- Generic replacements ---
// Standalone numeric IDs (4+ digits)
.replace(/\b\d{4,}\b/g, "<id>")
// File paths (Unix style)
.replace(/(?:\/[^\/\s]+){2,}/g, "<path>")
// File paths (Windows style)
.replace(/[A-Z]:\\(?:[^\\]+\\)+[^\\]+/g, "<path>")
// Email addresses
.replace(/\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b/g, "<email>")
// Memory addresses (0x...)
.replace(/0x[0-9a-fA-F]{8,}/g, "<addr>")
// Quoted strings with dynamic content
.replace(/"[^"]{20,}"/g, '"<string>"')
.replace(/'[^']{20,}'/g, "'<string>'")
);
}
/**
* Normalize stack trace by taking first few frames and removing dynamic parts.
*/
export function normalizeStackTrace(stack: string): string {
if (!stack) return "";
// Take first 5 stack frames only
const lines = stack.split("\n").slice(0, 5);
return lines
.map((line) => {
// Remove line and column numbers (file.ts:123:45 -> file.ts:_:_)
line = line.replace(/:\d+:\d+/g, ":_:_");
// Remove standalone numbers
line = line.replace(/\b\d+\b/g, "_");
// Remove file paths but keep filename
line = line.replace(/(?:\/[^\/\s]+)+\/([^\/\s]+)/g, "$1");
// Normalize whitespace
line = line.trim();
return line;
})
.filter((line) => line.length > 0)
.join("|");
}
+18 -1
View File
@@ -325,7 +325,7 @@ export function v3CreateBulkActionPath(
project: ProjectForPath,
environment: EnvironmentForPath,
filters?: TaskRunListSearchFilters,
mode?: "selected" | "filters",
mode?: "selected" | "filter",
action?: "replay" | "cancel"
) {
const searchParams = objectToSearchParams(filters) ?? new URLSearchParams();
@@ -527,6 +527,23 @@ export function v3LogsPath(
return `${v3EnvironmentPath(organization, project, environment)}/logs`;
}
export function v3ErrorsPath(
organization: OrgForPath,
project: ProjectForPath,
environment: EnvironmentForPath
) {
return `${v3EnvironmentPath(organization, project, environment)}/errors`;
}
export function v3ErrorPath(
organization: OrgForPath,
project: ProjectForPath,
environment: EnvironmentForPath,
error: { fingerprint: string }
) {
return `${v3ErrorsPath(organization, project, environment)}/${error.fingerprint}`;
}
export function v3DeploymentsPath(
organization: OrgForPath,
project: ProjectForPath,
+41
View File
@@ -0,0 +1,41 @@
/**
* Parses a version string into comparable numeric parts.
* Handles formats like "1.2.3", "20240115.1", "v1.0.0", plain timestamps, etc.
* Non-numeric pre-release suffixes (e.g. "-beta.1") are stripped for ordering purposes.
*/
function parseVersionParts(version: string): number[] {
const cleaned = version.replace(/^v/i, "").replace(/[-+].*$/, "");
return cleaned.split(".").map((p) => {
const n = parseInt(p, 10);
return isNaN(n) ? 0 : n;
});
}
/**
* Compares two version strings using numeric segment comparison (descending).
* Falls back to lexicographic comparison when segments are equal.
* Returns a negative number if `a` should come before `b` (i.e. `a` is newer).
*/
export function compareVersionsDescending(a: string, b: string): number {
const partsA = parseVersionParts(a);
const partsB = parseVersionParts(b);
const maxLen = Math.max(partsA.length, partsB.length);
for (let i = 0; i < maxLen; i++) {
const segA = partsA[i] ?? 0;
const segB = partsB[i] ?? 0;
if (segA !== segB) {
return segB - segA;
}
}
return b.localeCompare(a);
}
/**
* Sorts an array of version strings in descending order (newest first).
* Non-destructive returns a new array.
*/
export function sortVersionsDescending(versions: string[]): string[] {
return [...versions].sort(compareVersionsDescending);
}
+63
View File
@@ -0,0 +1,63 @@
import { z } from "zod";
import parseDuration from "parse-duration";
const DurationString = z
.string()
.refine(
(val) => {
const ms = parseDuration(val);
return ms !== null && ms > 0;
},
(val) => ({ message: `Invalid or non-positive duration string: "${val}"` })
);
const BracketSchema = z.object({
max: z.union([z.literal("Infinity"), DurationString]),
granularity: DurationString,
});
const BracketsSchema = z
.array(BracketSchema)
.min(1, "TimeGranularity requires at least one bracket");
export type TimeGranularityBracket = z.input<typeof BracketSchema>;
type ParsedBracket = {
maxMs: number;
granularityMs: number;
};
function requireParsedDuration(input: string): number {
const ms = parseDuration(input);
if (ms === null || ms <= 0) {
throw new Error(`Duration must be strictly positive, got "${input}" (${ms}ms)`);
}
return ms;
}
export class TimeGranularity {
private readonly parsed: ParsedBracket[];
constructor(brackets: TimeGranularityBracket[]) {
const validated = BracketsSchema.parse(brackets);
this.parsed = validated.map((b) => ({
maxMs: b.max === "Infinity" ? Infinity : requireParsedDuration(b.max),
granularityMs: requireParsedDuration(b.granularity),
}));
}
getTimeGranularityMs(from: Date, to: Date): number {
if (from.getTime() > to.getTime()) {
return this.parsed[this.parsed.length - 1].granularityMs;
}
const rangeMs = to.getTime() - from.getTime();
for (const bracket of this.parsed) {
if (rangeMs <= bracket.maxMs) {
return bracket.granularityMs;
}
}
return this.parsed[this.parsed.length - 1].granularityMs;
}
}
@@ -0,0 +1,410 @@
import { describe, it, expect } from "vitest";
import {
calculateErrorFingerprint,
normalizeErrorMessage,
normalizeStackTrace,
} from "~/utils/errorFingerprinting";
describe("normalizeErrorMessage", () => {
it("should normalize UUIDs", () => {
const message = "Error processing user 550e8400-e29b-41d4-a716-446655440000";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Error processing user <uuid>");
});
it("should normalize run IDs", () => {
const message = "Failed to execute run_abcd1234xyz";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Failed to execute <run-id>");
});
it("should normalize task friendly IDs", () => {
const message = "Task task_abc12345678 failed";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Task <id> failed");
});
it("should normalize numeric IDs (4+ digits)", () => {
const message = "User 12345 not found";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("User <id> not found");
});
it("should not normalize short numbers", () => {
const message = "Retry attempt 3 of 5";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Retry attempt 3 of 5");
});
it("should normalize ISO 8601 timestamps", () => {
const message = "Event at 2024-03-01T15:30:45Z failed";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Event at <timestamp> failed");
});
it("should normalize ISO timestamps with milliseconds", () => {
const message = "Timeout at 2024-03-01T15:30:45.123Z";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Timeout at <timestamp>");
});
it("should normalize Unix timestamps", () => {
const message = "Created at 1234567890";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Created at <timestamp>");
});
it("should normalize Unix timestamps (milliseconds)", () => {
const message = "Created at 1234567890123";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Created at <timestamp>");
});
it("should normalize Unix file paths", () => {
const message = "Cannot read /home/user/project/file.ts";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Cannot read <path>");
});
it("should normalize Windows file paths", () => {
const message = "Cannot read C:\\Users\\John\\project\\file.ts";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Cannot read <path>");
});
it("should normalize email addresses", () => {
const message = "Email user@example.com already exists";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Email <email> already exists");
});
it("should normalize URLs", () => {
const message = "Failed to fetch https://api.example.com/users/123";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Failed to fetch <url>");
});
it("should normalize HTTP URLs", () => {
const message = "Request to http://localhost:3000/api failed";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Request to <url> failed");
});
it("should normalize memory addresses", () => {
const message = "Segfault at 0x7fff5fbffab0";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Segfault at <addr>");
});
it("should normalize long quoted strings", () => {
const message = 'Error: "this is a very long error message with dynamic content that changes"';
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe('Error: "<string>"');
});
it("should handle multiple replacements", () => {
const message =
"User 12345 at user@example.com failed to access run_abc123 at 2024-03-01T15:30:45Z";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("User <id> at <email> failed to access <run-id> at <timestamp>");
});
it("should return empty string for empty input", () => {
expect(normalizeErrorMessage("")).toBe("");
});
it("should handle messages with no dynamic content", () => {
const message = "Connection timeout";
const normalized = normalizeErrorMessage(message);
expect(normalized).toBe("Connection timeout");
});
describe("ordering: specific patterns before generic ones", () => {
it("ISO timestamp year should not be consumed by numeric ID regex", () => {
const message = "Deadline was 2025-12-31T23:59:59Z";
expect(normalizeErrorMessage(message)).toBe("Deadline was <timestamp>");
});
it("ISO timestamp without trailing Z should normalize correctly", () => {
const message = "Started at 2024-01-15T08:00:00";
expect(normalizeErrorMessage(message)).toBe("Started at <timestamp>");
});
it("Unix timestamp (10 digits) should not become <id>", () => {
const message = "Token expires 1700000000";
expect(normalizeErrorMessage(message)).toBe("Token expires <timestamp>");
});
it("Unix timestamp (13 digits) should not become <id>", () => {
const message = "Sent at 1700000000000";
expect(normalizeErrorMessage(message)).toBe("Sent at <timestamp>");
});
it("URL path should not be stripped before URL regex runs", () => {
const message = "Webhook failed for https://hooks.example.com/webhook/abc";
expect(normalizeErrorMessage(message)).toBe("Webhook failed for <url>");
});
it("URL with port and path should normalize to <url>", () => {
const message = "Cannot reach http://localhost:8080/health/ready";
expect(normalizeErrorMessage(message)).toBe("Cannot reach <url>");
});
it("URL with query string should normalize to <url>", () => {
const message = "GET https://api.example.com/v2/users?page=1&limit=50 returned 500";
expect(normalizeErrorMessage(message)).toBe("GET <url> returned 500");
});
it("message with both a URL and a timestamp", () => {
const message =
"Request to https://api.example.com/data failed at 2025-06-15T10:30:00Z";
expect(normalizeErrorMessage(message)).toBe(
"Request to <url> failed at <timestamp>"
);
});
it("message with a URL and a unix timestamp", () => {
const message = "Callback to https://example.com/hook timed out after 1700000000";
expect(normalizeErrorMessage(message)).toBe(
"Callback to <url> timed out after <timestamp>"
);
});
it("path-like string that is NOT a URL should still become <path>", () => {
const message = "Cannot read /var/log/app/error.log";
expect(normalizeErrorMessage(message)).toBe("Cannot read <path>");
});
});
describe("fingerprint stability: same error class groups together despite dynamic values", () => {
it("errors differing only in ISO timestamp should share a fingerprint", () => {
const e1 = { type: "TimeoutError", message: "Timed out at 2025-01-01T00:00:00Z" };
const e2 = { type: "TimeoutError", message: "Timed out at 2026-06-15T12:30:00Z" };
expect(calculateErrorFingerprint(e1)).toBe(calculateErrorFingerprint(e2));
});
it("errors differing only in URL path should share a fingerprint", () => {
const e1 = {
type: "FetchError",
message: "Failed to fetch https://api.example.com/users/123",
};
const e2 = {
type: "FetchError",
message: "Failed to fetch https://api.example.com/orders/456",
};
expect(calculateErrorFingerprint(e1)).toBe(calculateErrorFingerprint(e2));
});
it("errors differing only in unix timestamp should share a fingerprint", () => {
const e1 = { type: "ExpiredError", message: "Token expired at 1700000000" };
const e2 = { type: "ExpiredError", message: "Token expired at 1800000000" };
expect(calculateErrorFingerprint(e1)).toBe(calculateErrorFingerprint(e2));
});
});
});
describe("normalizeStackTrace", () => {
it("should normalize line and column numbers", () => {
const stack = `Error: Test error
at functionName (file.ts:123:45)
at anotherFunction (other.ts:67:89)`;
const normalized = normalizeStackTrace(stack);
expect(normalized).toContain(":_:_");
expect(normalized).not.toContain(":123:45");
});
it("should remove standalone numbers", () => {
const stack = `Error: Test
at Object.<anonymous> (/path/to/file.ts:123:45)
at Module._compile (node:internal/modules/cjs/loader:456:78)`;
const normalized = normalizeStackTrace(stack);
expect(normalized).not.toMatch(/\b\d+\b/);
});
it("should keep only first 5 frames", () => {
const stack = `Error: Test
at frame1 (file1.ts:1:1)
at frame2 (file2.ts:2:2)
at frame3 (file3.ts:3:3)
at frame4 (file4.ts:4:4)
at frame5 (file5.ts:5:5)
at frame6 (file6.ts:6:6)
at frame7 (file7.ts:7:7)`;
const normalized = normalizeStackTrace(stack);
const frames = normalized.split("|");
expect(frames.length).toBeLessThanOrEqual(5);
});
it("should remove file paths but keep filenames", () => {
const stack = `Error: Test
at functionName (/home/user/project/src/file.ts:123:45)`;
const normalized = normalizeStackTrace(stack);
expect(normalized).toContain("file.ts");
expect(normalized).not.toContain("/home/user/project/src/");
});
it("should filter out empty lines", () => {
const stack = `Error: Test
at functionName (file.ts:123:45)
at anotherFunction (other.ts:67:89)`;
const normalized = normalizeStackTrace(stack);
const frames = normalized.split("|").filter((f) => f.length > 0);
expect(frames.length).toBeLessThanOrEqual(3);
});
it("should return empty string for empty stack", () => {
expect(normalizeStackTrace("")).toBe("");
});
it("should join frames with pipe delimiter", () => {
const stack = `Error: Test
at frame1 (file1.ts:1:1)
at frame2 (file2.ts:2:2)`;
const normalized = normalizeStackTrace(stack);
expect(normalized).toContain("|");
});
});
describe("calculateErrorFingerprint", () => {
it("should generate consistent fingerprints for same error", () => {
const error = {
type: "DatabaseError",
message: "Connection timeout",
stack: "at db.connect (db.ts:123:45)",
};
const fp1 = calculateErrorFingerprint(error);
const fp2 = calculateErrorFingerprint(error);
expect(fp1).toBe(fp2);
expect(fp1.length).toBe(16);
});
it("should generate same fingerprint for errors with different IDs", () => {
const error1 = {
type: "NotFoundError",
message: "User 12345 not found",
stack: "at findUser (user.ts:50:10)",
};
const error2 = {
type: "NotFoundError",
message: "User 67890 not found",
stack: "at findUser (user.ts:50:10)",
};
const fp1 = calculateErrorFingerprint(error1);
const fp2 = calculateErrorFingerprint(error2);
expect(fp1).toBe(fp2);
});
it("should generate same fingerprint for errors with different UUIDs", () => {
const error1 = {
type: "ValidationError",
message: "Invalid token 550e8400-e29b-41d4-a716-446655440000",
};
const error2 = {
type: "ValidationError",
message: "Invalid token 123e4567-e89b-12d3-a456-426614174000",
};
expect(calculateErrorFingerprint(error1)).toBe(calculateErrorFingerprint(error2));
});
it("should generate same fingerprint for errors with different run IDs", () => {
const error1 = {
type: "TaskError",
message: "Failed to execute run_abc123",
};
const error2 = {
type: "TaskError",
message: "Failed to execute run_xyz789",
};
expect(calculateErrorFingerprint(error1)).toBe(calculateErrorFingerprint(error2));
});
it("should generate different fingerprints for different error types", () => {
const error1 = {
type: "DatabaseError",
message: "Connection failed",
};
const error2 = {
type: "NetworkError",
message: "Connection failed",
};
expect(calculateErrorFingerprint(error1)).not.toBe(calculateErrorFingerprint(error2));
});
it("should generate different fingerprints for different error messages", () => {
const error1 = {
type: "Error",
message: "Connection timeout",
};
const error2 = {
type: "Error",
message: "Connection refused",
};
expect(calculateErrorFingerprint(error1)).not.toBe(calculateErrorFingerprint(error2));
});
it("should handle error with name instead of type", () => {
const error = {
name: "TypeError",
message: "Cannot read property 'foo' of undefined",
};
const fp = calculateErrorFingerprint(error);
expect(fp).toBeTruthy();
expect(fp.length).toBe(16);
});
it("should handle error with stacktrace instead of stack", () => {
const error = {
type: "Error",
message: "Test error",
stacktrace: "at test (file.ts:1:1)",
};
const fp = calculateErrorFingerprint(error);
expect(fp).toBeTruthy();
});
it("should return empty string for non-object error", () => {
expect(calculateErrorFingerprint(null)).toBe("");
expect(calculateErrorFingerprint(undefined)).toBe("");
expect(calculateErrorFingerprint("error string")).toBe("");
expect(calculateErrorFingerprint(123)).toBe("");
});
it("should handle errors with no message or stack", () => {
const error = {
type: "Error",
};
const fp = calculateErrorFingerprint(error);
expect(fp).toBeTruthy();
expect(fp.length).toBe(16);
});
it("should generate fingerprints using stack trace when available", () => {
const error1 = {
type: "Error",
message: "Test",
stack: "at funcA (a.ts:1:1)\nat funcB (b.ts:2:2)",
};
const error2 = {
type: "Error",
message: "Test",
stack: "at funcX (x.ts:1:1)\nat funcY (y.ts:2:2)",
};
expect(calculateErrorFingerprint(error1)).not.toBe(calculateErrorFingerprint(error2));
});
it("should normalize line numbers in stack traces for same code location", () => {
const error1 = {
type: "Error",
message: "Test",
stack: "at func (file.ts:123:45)",
};
const error2 = {
type: "Error",
message: "Test",
stack: "at func (file.ts:456:78)",
};
expect(calculateErrorFingerprint(error1)).toBe(calculateErrorFingerprint(error2));
});
});
@@ -0,0 +1,282 @@
# RunsReplicationService Error Fingerprinting Benchmark
This benchmark measures the performance impact of error fingerprinting in the RunsReplicationService.
## Overview
The benchmark:
1. Creates a realistic dataset of TaskRuns (7% with errors by default)
2. Runs the producer in a **separate process** to simulate real-world load
3. Measures replication throughput and Event Loop Utilization (ELU)
4. Compares performance with fingerprinting **enabled** vs **disabled**
## Architecture
```
┌─────────────────┐ ┌──────────────────────┐
│ Producer │ │ Benchmark Test │
│ (Child Process)│─────────│ (Main Process) │
│ │ IPC │ │
│ - Inserts │ │ - RunsReplication │
│ TaskRuns │ │ Service │
│ to Postgres │ │ - ELU Monitor │
│ │ │ - Metrics │
└─────────────────┘ └──────────────────────┘
│ │
│ │
▼ ▼
┌──────────┐ ┌──────────────┐
│ Postgres │ │ ClickHouse │
└──────────┘ └──────────────┘
```
## Files
- `runsReplicationBenchmark.test.ts` - Main benchmark test
- `runsReplicationBenchmark.producer.ts` - Producer script (runs in child process)
- `runsReplicationBenchmark.README.md` - This file
## Configuration
The benchmark can be configured via environment variables or by editing `BENCHMARK_CONFIG` in the test file:
```typescript
const BENCHMARK_CONFIG = {
// Number of runs to create
NUM_RUNS: parseInt(process.env.BENCHMARK_NUM_RUNS || "5000", 10),
// Error rate (0.07 = 7%)
ERROR_RATE: 0.07,
// Producer batch size
PRODUCER_BATCH_SIZE: 100,
// Replication service settings
FLUSH_BATCH_SIZE: 50,
FLUSH_INTERVAL_MS: 100,
MAX_FLUSH_CONCURRENCY: 4,
// Timeout
REPLICATION_TIMEOUT_MS: 120_000, // 2 minutes
};
```
## Running the Benchmark
### Quick Test (Small Dataset)
```bash
cd apps/webapp
BENCHMARK_NUM_RUNS=1000 pnpm run test ./test/runsReplicationBenchmark.test.ts --run
```
### Realistic Benchmark (Larger Dataset)
```bash
cd apps/webapp
BENCHMARK_NUM_RUNS=10000 pnpm run test ./test/runsReplicationBenchmark.test.ts --run
```
### High Volume Benchmark
```bash
cd apps/webapp
BENCHMARK_NUM_RUNS=50000 pnpm run test ./test/runsReplicationBenchmark.test.ts --run
```
**Note:** The benchmark is gated by the `BENCHMARKS_ENABLED` environment variable
(via `containerTest.skipIf`), so you don't need to edit the test file. Set
`BENCHMARKS_ENABLED=1` (and optionally `BENCHMARK_NUM_RUNS`) then run:
```bash
cd apps/webapp
BENCHMARKS_ENABLED=1 pnpm run test ./test/runsReplicationBenchmark.test.ts --run
```
## What Gets Measured
### 1. Producer Metrics
- Total runs created
- Runs with errors (should be ~7%)
- Duration
- Throughput (runs/sec)
### 2. Replication Metrics
- Total runs replicated to ClickHouse
- Replication duration
- Replication throughput (runs/sec)
### 3. Event Loop Utilization (ELU)
- Mean utilization (%)
- P50 (median) utilization (%)
- P95 utilization (%)
- P99 utilization (%)
- All samples for detailed analysis
### 4. OpenTelemetry Metrics
- Batches flushed
- Task runs inserted
- Payloads inserted
- Events processed
## Output
The benchmark produces detailed output including:
```
================================================================================
BENCHMARK: baseline-no-fingerprinting
Error Fingerprinting: DISABLED
Runs: 5000, Error Rate: 7.0%
================================================================================
[Producer] Starting - will create 5000 runs (7.0% with errors)
[Producer] Progress: 1000/5000 runs (2500 runs/sec)
...
[Producer] Completed:
- Total runs: 5000
- With errors: 352 (7.0%)
- Duration: 2145ms
- Throughput: 2331 runs/sec
[Benchmark] Waiting for replication to complete...
================================================================================
RESULTS: baseline-no-fingerprinting
================================================================================
Producer:
Created: 5000 runs
With errors: 352 (7.0%)
Duration: 2145ms
Throughput: 2331 runs/sec
Replication:
Replicated: 5000 runs
Duration: 3456ms
Throughput: 1447 runs/sec
Event Loop Utilization:
Mean: 23.45%
P50: 22.10%
P95: 34.20%
P99: 41.30%
Samples: 346
Metrics:
Batches flushed: 102
Task runs inserted: 5000
Payloads inserted: 5000
Events processed: 5000
================================================================================
[... Similar output for "with-fingerprinting" benchmark ...]
================================================================================
COMPARISON
Baseline: baseline-no-fingerprinting (fingerprinting OFF)
Comparison: with-fingerprinting (fingerprinting ON)
================================================================================
Replication Duration:
3456ms → 3512ms (+1.62%)
Throughput:
1447 → 1424 runs/sec (-1.59%)
Event Loop Utilization (Mean):
23.45% → 24.12% (+2.86%)
Event Loop Utilization (P99):
41.30% → 43.20% (+4.60%)
================================================================================
BENCHMARK COMPLETE
Fingerprinting impact on replication duration: +1.62%
Fingerprinting impact on throughput: -1.59%
Fingerprinting impact on ELU (mean): +2.86%
Fingerprinting impact on ELU (P99): +4.60%
```
## Interpreting Results
### What to Look For
1. **Replication Duration Delta** - How much longer replication takes with fingerprinting
2. **Throughput Delta** - Change in runs processed per second
3. **ELU Delta** - Change in event loop utilization (higher = more CPU bound)
### Expected Results
With a 7% error rate and SHA-256 hashing:
- **Small impact** (<5% overhead): Fingerprinting is well optimized
- **Moderate impact** (5-15% overhead): May want to consider optimizations
- **Large impact** (>15% overhead): Fingerprinting needs optimization
### Performance Optimization Ideas
If the benchmark shows significant overhead, consider:
1. **Faster hashing algorithm** - Replace SHA-256 with xxHash or MurmurHash3
2. **Worker threads** - Move fingerprinting to worker threads
3. **Caching** - Cache fingerprints for identical errors
4. **Lazy computation** - Only compute fingerprints when needed
5. **Batch processing** - Group similar errors before hashing
## Dataset Characteristics
The producer generates realistic error variety:
- TypeError (undefined property access)
- Error (API fetch failures)
- ValidationError (input validation)
- TimeoutError (operation timeouts)
- DatabaseError (connection failures)
- ReferenceError (undefined variables)
Each error template includes:
- Realistic stack traces
- Variable IDs and timestamps
- Line/column numbers
- File paths
This ensures the fingerprinting algorithm is tested with realistic data.
## Troubleshooting
### Benchmark Times Out
Increase the timeout:
```typescript
REPLICATION_TIMEOUT_MS: 300_000, // 5 minutes
```
### Producer Fails
Check Postgres connection and ensure:
- Docker services are running (`pnpm run docker`)
- Database is accessible
- Sufficient disk space
### Different Results Each Run
This is normal! Factors affecting variance:
- System load
- Docker container overhead
- Database I/O
- Network latency (even localhost)
Run multiple times and look at trends.
## Future Enhancements
Potential improvements to the benchmark:
1. **Multiple error rates** - Test 0%, 5%, 10%, 25%, 50% error rates
2. **Different hash algorithms** - Compare SHA-256 vs xxHash vs MurmurHash3
3. **Worker thread comparison** - Test main thread vs worker threads
4. **Concurrent producers** - Multiple producer processes
5. **Memory profiling** - Track memory usage over time
6. **Flame graphs** - Generate CPU flame graphs for analysis
7. **Historical tracking** - Store results over time to track regressions
@@ -0,0 +1,212 @@
#!/usr/bin/env node
/**
* Producer script that runs in a separate process to insert TaskRuns into PostgreSQL.
* This simulates realistic production load for benchmarking RunsReplicationService.
*/
import { PrismaClient } from "@trigger.dev/database";
import { performance } from "node:perf_hooks";
interface ProducerConfig {
postgresUrl: string;
organizationId: string;
projectId: string;
environmentId: string;
numRuns: number;
errorRate: number; // 0.07 = 7%
batchSize: number;
}
// Error templates for realistic variety
const ERROR_TEMPLATES = [
{
type: "TypeError",
message: "Cannot read property 'foo' of undefined",
stack: `TypeError: Cannot read property 'foo' of undefined
at processData (/app/src/handler.ts:42:15)
at runTask (/app/src/runtime.ts:128:20)
at executeRun (/app/src/executor.ts:89:12)
at async Runner.execute (/app/src/runner.ts:56:5)`,
},
{
type: "Error",
message: "Failed to fetch data from API endpoint https://api.example.com/data/12345",
stack: `Error: Failed to fetch data from API endpoint https://api.example.com/data/12345
at fetchData (/app/src/api.ts:78:11)
at getData (/app/src/service.ts:34:18)
at processTask (/app/src/handler.ts:23:15)
at runTask (/app/src/runtime.ts:128:20)`,
},
{
type: "ValidationError",
message: "Invalid input: expected string for field 'email', got number: 1234567890",
stack: `ValidationError: Invalid input: expected string for field 'email', got number: 1234567890
at validateInput (/app/src/validator.ts:156:9)
at processRequest (/app/src/handler.ts:67:23)
at runTask (/app/src/runtime.ts:128:20)`,
},
{
type: "TimeoutError",
message: "Operation timed out after 30000ms",
stack: `TimeoutError: Operation timed out after 30000ms
at Timeout._onTimeout (/app/src/timeout.ts:45:15)
at processTask (/app/src/handler.ts:89:12)
at runTask (/app/src/runtime.ts:128:20)`,
},
{
type: "DatabaseError",
message: "Connection to database 'prod_db' failed: timeout of 5000ms exceeded",
stack: `DatabaseError: Connection to database 'prod_db' failed: timeout of 5000ms exceeded
at connect (/app/node_modules/pg/lib/client.js:234:11)
at query (/app/src/db.ts:89:18)
at getData (/app/src/service.ts:45:22)`,
},
{
type: "ReferenceError",
message: "userId is not defined",
stack: `ReferenceError: userId is not defined
at validateUser (/app/src/auth.ts:123:9)
at processTask (/app/src/handler.ts:34:15)
at runTask (/app/src/runtime.ts:128:20)`,
},
];
function generateError() {
const template = ERROR_TEMPLATES[Math.floor(Math.random() * ERROR_TEMPLATES.length)];
// Add variation to make errors slightly different
const randomId = Math.floor(Math.random() * 100000);
const randomTimestamp = Date.now() + Math.floor(Math.random() * 10000);
return {
type: template.type,
name: template.type,
message: template.message
.replace(/\d{4,}/g, String(randomId))
.replace(/\d{13}/g, String(randomTimestamp)),
stack: template.stack
.replace(/:\d+:\d+/g, `:${Math.floor(Math.random() * 500)}:${Math.floor(Math.random() * 50)}`)
.replace(/\d{4,}/g, String(randomId)),
};
}
async function runProducer(config: ProducerConfig) {
const prisma = new PrismaClient({
datasources: {
db: {
url: config.postgresUrl,
},
},
});
try {
console.log(
`[Producer] Starting - will create ${config.numRuns} runs (${(config.errorRate * 100).toFixed(
1
)}% with errors)`
);
const startTime = performance.now();
let created = 0;
let withErrors = 0;
// Process in batches to avoid overwhelming the database
for (let batch = 0; batch < Math.ceil(config.numRuns / config.batchSize); batch++) {
const batchStart = batch * config.batchSize;
const batchEnd = Math.min(batchStart + config.batchSize, config.numRuns);
const batchSize = batchEnd - batchStart;
const runs = [];
for (let i = batchStart; i < batchEnd; i++) {
const hasError = Math.random() < config.errorRate;
const status = hasError ? "COMPLETED_WITH_ERRORS" : "COMPLETED_SUCCESSFULLY";
const runData: any = {
friendlyId: `run_bench_${Date.now()}_${i}`,
taskIdentifier: `benchmark-task-${i % 10}`, // Vary task identifiers
payload: JSON.stringify({ index: i, timestamp: Date.now() }),
traceId: `trace_${i}`,
spanId: `span_${i}`,
queue: `queue-${i % 5}`, // Vary queues
runtimeEnvironmentId: config.environmentId,
projectId: config.projectId,
organizationId: config.organizationId,
environmentType: "DEVELOPMENT",
engine: "V2",
status,
createdAt: new Date(Date.now() - Math.floor(Math.random() * 1000)),
updatedAt: new Date(),
};
if (hasError) {
runData.error = generateError();
withErrors++;
}
runs.push(runData);
}
// Insert batch
await prisma.taskRun.createMany({
data: runs,
});
created += batchSize;
if (batch % 10 === 0 || batch === Math.ceil(config.numRuns / config.batchSize) - 1) {
const elapsed = performance.now() - startTime;
const rate = (created / elapsed) * 1000;
console.log(
`[Producer] Progress: ${created}/${config.numRuns} runs (${rate.toFixed(0)} runs/sec)`
);
}
}
const endTime = performance.now();
const duration = endTime - startTime;
const throughput = (created / duration) * 1000;
console.log(`[Producer] Completed:`);
console.log(` - Total runs: ${created}`);
console.log(` - With errors: ${withErrors} (${((withErrors / created) * 100).toFixed(1)}%)`);
console.log(` - Duration: ${duration.toFixed(0)}ms`);
console.log(` - Throughput: ${throughput.toFixed(0)} runs/sec`);
// Send results to parent process
if (process.send) {
process.send({
type: "complete",
stats: {
created,
withErrors,
duration,
throughput,
},
});
}
} catch (error) {
console.error("[Producer] Error:", error);
if (process.send) {
process.send({
type: "error",
error: error instanceof Error ? error.message : String(error),
});
}
process.exit(1);
} finally {
await prisma.$disconnect();
}
}
// Parse config from command line args
const configArg = process.argv[2];
if (!configArg) {
console.error("Usage: runsReplicationBenchmark.producer.ts <config-json>");
process.exit(1);
}
// This is ok for a benchmark script, but not for production code.
const config = JSON.parse(configArg) as ProducerConfig;
runProducer(config).catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});
@@ -0,0 +1,570 @@
import { ClickHouse } from "@internal/clickhouse";
import { containerTest } from "@internal/testcontainers";
import { fork, type ChildProcess } from "node:child_process";
import { performance, PerformanceObserver } from "node:perf_hooks";
import { setTimeout } from "node:timers/promises";
import path from "node:path";
import { z } from "zod";
import { RunsReplicationService } from "~/services/runsReplicationService.server";
import { createInMemoryTracing, createInMemoryMetrics } from "./utils/tracing";
// Extend test timeout for benchmarks
vi.setConfig({ testTimeout: 300_000 }); // 5 minutes
/**
* Benchmark configuration
*/
const BENCHMARK_CONFIG = {
// Number of runs to create - adjust this to test different volumes
// Start with smaller numbers (1000) for quick tests, increase to 10000+ for realistic benchmarks
NUM_RUNS: parseInt(process.env.BENCHMARK_NUM_RUNS || "5000", 10),
// Error rate (7% = realistic production load with some failures)
ERROR_RATE: 0.07,
// Batch size for producer
PRODUCER_BATCH_SIZE: 100,
// Replication service settings
FLUSH_BATCH_SIZE: 50,
FLUSH_INTERVAL_MS: 100,
MAX_FLUSH_CONCURRENCY: 4,
// How long to wait for replication to complete (in ms)
REPLICATION_TIMEOUT_MS: 120_000, // 2 minutes
};
interface BenchmarkResult {
name: string;
fingerprintingEnabled: boolean;
producerStats: {
created: number;
withErrors: number;
duration: number;
throughput: number;
};
replicationStats: {
duration: number;
throughput: number;
replicatedRuns: number;
};
eluStats: {
mean: number;
p50: number;
p95: number;
p99: number;
samples: number[];
};
metricsStats: {
batchesFlushed: number;
taskRunsInserted: number;
payloadsInserted: number;
eventsProcessed: number;
};
}
/**
* Measure Event Loop Utilization during benchmark
*/
class ELUMonitor {
private samples: number[] = [];
private interval: NodeJS.Timeout | null = null;
private startELU: { idle: number; active: number } | null = null;
start(intervalMs: number = 100) {
this.samples = [];
this.startELU = performance.eventLoopUtilization();
this.interval = setInterval(() => {
const elu = performance.eventLoopUtilization();
const utilization = elu.utilization * 100; // Convert to percentage
this.samples.push(utilization);
}, intervalMs);
}
stop(): { mean: number; p50: number; p95: number; p99: number; samples: number[] } {
if (this.interval) {
clearInterval(this.interval);
this.interval = null;
}
if (this.samples.length === 0) {
return { mean: 0, p50: 0, p95: 0, p99: 0, samples: [] };
}
const sorted = [...this.samples].sort((a, b) => a - b);
const mean = sorted.reduce((sum, val) => sum + val, 0) / sorted.length;
const p50 = sorted[Math.floor(sorted.length * 0.5)];
const p95 = sorted[Math.floor(sorted.length * 0.95)];
const p99 = sorted[Math.floor(sorted.length * 0.99)];
return { mean, p50, p95, p99, samples: sorted };
}
}
/**
* Run the producer script in a separate process
*/
async function runProducer(config: {
postgresUrl: string;
organizationId: string;
projectId: string;
environmentId: string;
numRuns: number;
errorRate: number;
batchSize: number;
}): Promise<{ created: number; withErrors: number; duration: number; throughput: number }> {
return new Promise((resolve, reject) => {
const producerPath = path.join(__dirname, "runsReplicationBenchmark.producer.ts");
// Use tsx to run the TypeScript file directly
const child = fork(producerPath, [JSON.stringify(config)], {
stdio: ["ignore", "pipe", "pipe", "ipc"],
execArgv: ["-r", "tsx/cjs"],
});
let output = "";
child.stdout?.on("data", (data) => {
const text = data.toString();
output += text;
console.log(text.trim());
});
child.stderr?.on("data", (data) => {
console.error(data.toString().trim());
});
child.on("message", (message: any) => {
if (message.type === "complete") {
resolve(message.stats);
} else if (message.type === "error") {
reject(new Error(message.error));
}
});
child.on("error", (error) => {
reject(error);
});
child.on("exit", (code) => {
if (code !== 0) {
reject(new Error(`Producer exited with code ${code}`));
}
});
});
}
/**
* Wait for all runs to be replicated to ClickHouse
*/
async function waitForReplication(
clickhouse: ClickHouse,
organizationId: string,
expectedCount: number,
timeoutMs: number
): Promise<{ duration: number; replicatedRuns: number }> {
const startTime = performance.now();
const deadline = startTime + timeoutMs;
const queryRuns = clickhouse.reader.query({
name: "benchmark-count",
query:
"SELECT count(*) as count FROM trigger_dev.task_runs_v2 WHERE organization_id = {org_id:String}",
schema: z.object({ count: z.number() }),
params: z.object({ org_id: z.string() }),
});
while (performance.now() < deadline) {
const [error, result] = await queryRuns({ org_id: organizationId });
if (error) {
throw new Error(`Failed to query ClickHouse: ${error.message}`);
}
const count = result?.[0]?.count || 0;
if (count >= expectedCount) {
const duration = performance.now() - startTime;
return { duration, replicatedRuns: count };
}
// Wait a bit before checking again
await setTimeout(500);
}
throw new Error(
`Replication timeout: expected ${expectedCount} runs, but only found ${await getRunCount(
clickhouse
)} after ${timeoutMs}ms`
);
}
async function getRunCount(clickhouse: ClickHouse): Promise<number> {
const queryRuns = clickhouse.reader.query({
name: "benchmark-count",
query: "SELECT count(*) as count FROM trigger_dev.task_runs_v2",
schema: z.object({ count: z.number() }),
});
const [error, result] = await queryRuns({});
if (error) return 0;
return result?.[0]?.count || 0;
}
/**
* Extract metrics from OpenTelemetry metrics
*/
function extractMetrics(metrics: any[]): {
batchesFlushed: number;
taskRunsInserted: number;
payloadsInserted: number;
eventsProcessed: number;
} {
function getMetricData(name: string) {
for (const resourceMetrics of metrics) {
for (const scopeMetrics of resourceMetrics.scopeMetrics) {
for (const metric of scopeMetrics.metrics) {
if (metric.descriptor.name === name) {
return metric;
}
}
}
}
return null;
}
function sumCounterValues(metric: any): number {
if (!metric?.dataPoints) return 0;
return metric.dataPoints.reduce((sum: number, dp: any) => sum + (dp.value || 0), 0);
}
return {
batchesFlushed: sumCounterValues(getMetricData("runs_replication.batches_flushed")),
taskRunsInserted: sumCounterValues(getMetricData("runs_replication.task_runs_inserted")),
payloadsInserted: sumCounterValues(getMetricData("runs_replication.payloads_inserted")),
eventsProcessed: sumCounterValues(getMetricData("runs_replication.events_processed")),
};
}
/**
* Run a single benchmark test
*/
async function runBenchmark(
name: string,
fingerprintingEnabled: boolean,
{
clickhouseContainer,
redisOptions,
postgresContainer,
prisma,
}: {
clickhouseContainer: any;
redisOptions: any;
postgresContainer: any;
prisma: any;
}
): Promise<BenchmarkResult> {
console.log(`\n${"=".repeat(80)}`);
console.log(`BENCHMARK: ${name}`);
console.log(`Error Fingerprinting: ${fingerprintingEnabled ? "ENABLED" : "DISABLED"}`);
console.log(
`Runs: ${BENCHMARK_CONFIG.NUM_RUNS}, Error Rate: ${(BENCHMARK_CONFIG.ERROR_RATE * 100).toFixed(
1
)}%`
);
console.log(`${"=".repeat(80)}\n`);
// Setup
const organization = await prisma.organization.create({
data: {
title: `benchmark-${name}`,
slug: `benchmark-${name}`,
},
});
const project = await prisma.project.create({
data: {
name: `benchmark-${name}`,
slug: `benchmark-${name}`,
organizationId: organization.id,
externalRef: `benchmark-${name}`,
},
});
const runtimeEnvironment = await prisma.runtimeEnvironment.create({
data: {
slug: `benchmark-${name}`,
type: "DEVELOPMENT",
projectId: project.id,
organizationId: organization.id,
apiKey: `benchmark-${name}`,
pkApiKey: `benchmark-${name}`,
shortcode: `benchmark-${name}`,
},
});
// Setup ClickHouse
const clickhouse = new ClickHouse({
url: clickhouseContainer.getConnectionUrl(),
name: `benchmark-${name}`,
compression: {
request: true,
},
logLevel: "warn",
});
// Setup tracing and metrics
const { tracer } = createInMemoryTracing();
const metricsHelper = createInMemoryMetrics();
// Create and start replication service
const runsReplicationService = new RunsReplicationService({
clickhouse,
pgConnectionUrl: postgresContainer.getConnectionUri(),
serviceName: `benchmark-${name}`,
slotName: `benchmark_${name.replace(/-/g, "_")}`,
publicationName: `benchmark_${name.replace(/-/g, "_")}_pub`,
redisOptions,
maxFlushConcurrency: BENCHMARK_CONFIG.MAX_FLUSH_CONCURRENCY,
flushIntervalMs: BENCHMARK_CONFIG.FLUSH_INTERVAL_MS,
flushBatchSize: BENCHMARK_CONFIG.FLUSH_BATCH_SIZE,
leaderLockTimeoutMs: 10000,
leaderLockExtendIntervalMs: 2000,
ackIntervalSeconds: 10,
tracer,
meter: metricsHelper.meter,
logLevel: "warn",
disableErrorFingerprinting: !fingerprintingEnabled,
});
await runsReplicationService.start();
// Start ELU monitoring
const eluMonitor = new ELUMonitor();
eluMonitor.start(100);
let producerStats!: BenchmarkResult["producerStats"];
let replicationResult!: { duration: number; replicatedRuns: number };
let metricsStats!: BenchmarkResult["metricsStats"];
let eluStats!: BenchmarkResult["eluStats"];
try {
// Run producer in separate process
console.log("\n[Benchmark] Starting producer...");
producerStats = await runProducer({
postgresUrl: postgresContainer.getConnectionUri(),
organizationId: organization.id,
projectId: project.id,
environmentId: runtimeEnvironment.id,
numRuns: BENCHMARK_CONFIG.NUM_RUNS,
errorRate: BENCHMARK_CONFIG.ERROR_RATE,
batchSize: BENCHMARK_CONFIG.PRODUCER_BATCH_SIZE,
});
console.log("\n[Benchmark] Waiting for replication to complete...");
replicationResult = await waitForReplication(
clickhouse,
organization.id,
producerStats.created,
BENCHMARK_CONFIG.REPLICATION_TIMEOUT_MS
);
const metrics = await metricsHelper.getMetrics();
metricsStats = extractMetrics(metrics);
} finally {
eluStats = eluMonitor.stop();
await runsReplicationService.stop();
await metricsHelper.shutdown();
}
const throughput = (replicationResult.replicatedRuns / replicationResult.duration) * 1000;
const result: BenchmarkResult = {
name,
fingerprintingEnabled,
producerStats,
replicationStats: {
duration: replicationResult.duration,
throughput,
replicatedRuns: replicationResult.replicatedRuns,
},
eluStats,
metricsStats,
};
// Print results
console.log(`\n${"=".repeat(80)}`);
console.log(`RESULTS: ${name}`);
console.log(`${"=".repeat(80)}`);
console.log("\nProducer:");
console.log(` Created: ${producerStats.created} runs`);
console.log(
` With errors: ${producerStats.withErrors} (${(
(producerStats.withErrors / producerStats.created) *
100
).toFixed(1)}%)`
);
console.log(` Duration: ${producerStats.duration.toFixed(0)}ms`);
console.log(` Throughput: ${producerStats.throughput.toFixed(0)} runs/sec`);
console.log("\nReplication:");
console.log(` Replicated: ${replicationResult.replicatedRuns} runs`);
console.log(` Duration: ${replicationResult.duration.toFixed(0)}ms`);
console.log(` Throughput: ${throughput.toFixed(0)} runs/sec`);
console.log("\nEvent Loop Utilization:");
console.log(` Mean: ${eluStats.mean.toFixed(2)}%`);
console.log(` P50: ${eluStats.p50.toFixed(2)}%`);
console.log(` P95: ${eluStats.p95.toFixed(2)}%`);
console.log(` P99: ${eluStats.p99.toFixed(2)}%`);
console.log(` Samples: ${eluStats.samples.length}`);
console.log("\nMetrics:");
console.log(` Batches flushed: ${metricsStats.batchesFlushed}`);
console.log(` Task runs inserted: ${metricsStats.taskRunsInserted}`);
console.log(` Payloads inserted: ${metricsStats.payloadsInserted}`);
console.log(` Events processed: ${metricsStats.eventsProcessed}`);
console.log(`${"=".repeat(80)}\n`);
return result;
}
/**
* Compare two benchmark results and print delta
*/
function compareBenchmarks(baseline: BenchmarkResult, comparison: BenchmarkResult) {
console.log(`\n${"=".repeat(80)}`);
console.log("COMPARISON");
console.log(
`Baseline: ${baseline.name} (fingerprinting ${baseline.fingerprintingEnabled ? "ON" : "OFF"})`
);
console.log(
`Comparison: ${comparison.name} (fingerprinting ${
comparison.fingerprintingEnabled ? "ON" : "OFF"
})`
);
console.log(`${"=".repeat(80)}`);
const replicationDurationDelta =
((comparison.replicationStats.duration - baseline.replicationStats.duration) /
baseline.replicationStats.duration) *
100;
const throughputDelta =
((comparison.replicationStats.throughput - baseline.replicationStats.throughput) /
baseline.replicationStats.throughput) *
100;
const eluMeanDelta =
((comparison.eluStats.mean - baseline.eluStats.mean) / baseline.eluStats.mean) * 100;
const eluP99Delta =
((comparison.eluStats.p99 - baseline.eluStats.p99) / baseline.eluStats.p99) * 100;
console.log("\nReplication Duration:");
console.log(
` ${baseline.replicationStats.duration.toFixed(
0
)}ms → ${comparison.replicationStats.duration.toFixed(0)}ms (${
replicationDurationDelta > 0 ? "+" : ""
}${replicationDurationDelta.toFixed(2)}%)`
);
console.log("\nThroughput:");
console.log(
` ${baseline.replicationStats.throughput.toFixed(
0
)}${comparison.replicationStats.throughput.toFixed(0)} runs/sec (${
throughputDelta > 0 ? "+" : ""
}${throughputDelta.toFixed(2)}%)`
);
console.log("\nEvent Loop Utilization (Mean):");
console.log(
` ${baseline.eluStats.mean.toFixed(2)}% → ${comparison.eluStats.mean.toFixed(2)}% (${
eluMeanDelta > 0 ? "+" : ""
}${eluMeanDelta.toFixed(2)}%)`
);
console.log("\nEvent Loop Utilization (P99):");
console.log(
` ${baseline.eluStats.p99.toFixed(2)}% → ${comparison.eluStats.p99.toFixed(2)}% (${
eluP99Delta > 0 ? "+" : ""
}${eluP99Delta.toFixed(2)}%)`
);
console.log(`\n${"=".repeat(80)}\n`);
// Return deltas for assertions if needed
return {
replicationDurationDelta,
throughputDelta,
eluMeanDelta,
eluP99Delta,
};
}
describe("RunsReplicationService Benchmark", () => {
containerTest.skipIf(process.env.BENCHMARKS_ENABLED !== "1")(
"should benchmark error fingerprinting performance impact",
async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => {
// Enable replica identity for TaskRun table
await prisma.$executeRawUnsafe(`ALTER TABLE public."TaskRun" REPLICA IDENTITY FULL;`);
console.log("\n" + "=".repeat(80));
console.log("RUNS REPLICATION SERVICE - ERROR FINGERPRINTING BENCHMARK");
console.log("=".repeat(80));
console.log(`Configuration:`);
console.log(` Total runs: ${BENCHMARK_CONFIG.NUM_RUNS}`);
console.log(` Error rate: ${(BENCHMARK_CONFIG.ERROR_RATE * 100).toFixed(1)}%`);
console.log(
` Expected errors: ~${Math.floor(BENCHMARK_CONFIG.NUM_RUNS * BENCHMARK_CONFIG.ERROR_RATE)}`
);
console.log(` Producer batch size: ${BENCHMARK_CONFIG.PRODUCER_BATCH_SIZE}`);
console.log(` Replication batch size: ${BENCHMARK_CONFIG.FLUSH_BATCH_SIZE}`);
console.log(` Max flush concurrency: ${BENCHMARK_CONFIG.MAX_FLUSH_CONCURRENCY}`);
console.log("=".repeat(80) + "\n");
// Run benchmark WITHOUT error fingerprinting (baseline)
const baselineResult = await runBenchmark("baseline-no-fingerprinting", false, {
clickhouseContainer,
redisOptions,
postgresContainer,
prisma,
});
// Run benchmark WITH error fingerprinting
const fingerprintingResult = await runBenchmark("with-fingerprinting", true, {
clickhouseContainer,
redisOptions,
postgresContainer,
prisma,
});
// Compare results
const deltas = compareBenchmarks(baselineResult, fingerprintingResult);
// Basic assertions - just to ensure benchmarks completed successfully
expect(baselineResult.replicationStats.replicatedRuns).toBe(BENCHMARK_CONFIG.NUM_RUNS);
expect(fingerprintingResult.replicationStats.replicatedRuns).toBe(BENCHMARK_CONFIG.NUM_RUNS);
// Log final summary
console.log("BENCHMARK COMPLETE");
console.log(
`Fingerprinting impact on replication duration: ${
deltas.replicationDurationDelta > 0 ? "+" : ""
}${deltas.replicationDurationDelta.toFixed(2)}%`
);
console.log(
`Fingerprinting impact on throughput: ${
deltas.throughputDelta > 0 ? "+" : ""
}${deltas.throughputDelta.toFixed(2)}%`
);
console.log(
`Fingerprinting impact on ELU (mean): ${
deltas.eluMeanDelta > 0 ? "+" : ""
}${deltas.eluMeanDelta.toFixed(2)}%`
);
console.log(
`Fingerprinting impact on ELU (P99): ${
deltas.eluP99Delta > 0 ? "+" : ""
}${deltas.eluP99Delta.toFixed(2)}%`
);
}
);
});
+60
View File
@@ -0,0 +1,60 @@
import { describe, expect, it } from "vitest";
import { TimeGranularity } from "~/utils/timeGranularity";
const SECOND = 1_000;
const MINUTE = 60 * SECOND;
const HOUR = 60 * MINUTE;
function makeRange(durationMs: number): [Date, Date] {
const from = new Date("2025-01-01T00:00:00Z");
const to = new Date(from.getTime() + durationMs);
return [from, to];
}
describe("TimeGranularity", () => {
const granularity = new TimeGranularity([
{ max: "1h", granularity: "10s" },
{ max: "6h", granularity: "1m" },
{ max: "Infinity", granularity: "10m" },
]);
it("returns the first bracket when range is within its max", () => {
const [from, to] = makeRange(30 * MINUTE);
expect(granularity.getTimeGranularityMs(from, to)).toBe(10 * SECOND);
});
it("returns a middle bracket when range exceeds the first but not the second", () => {
const [from, to] = makeRange(2 * HOUR);
expect(granularity.getTimeGranularityMs(from, to)).toBe(1 * MINUTE);
});
it("returns the last bracket when range exceeds all non-Infinity maxes", () => {
const [from, to] = makeRange(24 * HOUR);
expect(granularity.getTimeGranularityMs(from, to)).toBe(10 * MINUTE);
});
it("matches a bracket when range exactly equals its max", () => {
const [from, to] = makeRange(1 * HOUR);
expect(granularity.getTimeGranularityMs(from, to)).toBe(10 * SECOND);
});
it("moves to the next bracket when range exceeds a boundary by 1ms", () => {
const [from, to] = makeRange(1 * HOUR + 1);
expect(granularity.getTimeGranularityMs(from, to)).toBe(1 * MINUTE);
});
it("returns the first bracket's granularity for a zero-length range", () => {
const date = new Date("2025-01-01T00:00:00Z");
expect(granularity.getTimeGranularityMs(date, date)).toBe(10 * SECOND);
});
it("returns the broadest granularity for an inverted range (from > to)", () => {
const from = new Date("2025-01-01T01:00:00Z");
const to = new Date("2025-01-01T00:00:00Z");
expect(granularity.getTimeGranularityMs(from, to)).toBe(10 * MINUTE);
});
it("throws when constructed with an empty array", () => {
expect(() => new TimeGranularity([])).toThrow("at least one bracket");
});
});
@@ -0,0 +1,11 @@
-- +goose Up
ALTER TABLE trigger_dev.task_runs_v2
ADD COLUMN error_fingerprint String DEFAULT '';
-- Bloom filter index for fast error fingerprint lookups
ALTER TABLE trigger_dev.task_runs_v2
ADD INDEX idx_error_fingerprint error_fingerprint TYPE bloom_filter GRANULARITY 4;
-- +goose Down
ALTER TABLE trigger_dev.task_runs_v2 DROP INDEX idx_error_fingerprint;
ALTER TABLE trigger_dev.task_runs_v2 DROP COLUMN error_fingerprint;
@@ -0,0 +1,78 @@
-- +goose Up
-- Aggregated error groups table (per task + fingerprint)
CREATE TABLE trigger_dev.errors_v1
(
organization_id String,
project_id String,
environment_id String,
task_identifier String,
error_fingerprint String,
-- Error details (samples from occurrences)
error_type String,
error_message String,
sample_stack_trace String,
-- SimpleAggregateFunction stores raw values and applies the function during merge,
-- avoiding binary state encoding issues with AggregateFunction.
last_seen_date SimpleAggregateFunction(max, DateTime),
first_seen SimpleAggregateFunction(min, DateTime64(3)),
last_seen SimpleAggregateFunction(max, DateTime64(3)),
occurrence_count AggregateFunction(sum, UInt64),
affected_task_versions AggregateFunction(uniq, String),
-- Samples for debugging
sample_run_id AggregateFunction(any, String),
sample_friendly_id AggregateFunction(any, String),
-- Status distribution
status_distribution AggregateFunction(sumMap, Array(String), Array(UInt64))
)
ENGINE = AggregatingMergeTree()
ORDER BY (organization_id, project_id, environment_id, task_identifier, error_fingerprint)
TTL last_seen_date + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;
-- Materialized view to auto-populate from task_runs_v2
CREATE MATERIALIZED VIEW trigger_dev.errors_mv_v1
TO trigger_dev.errors_v1
AS
SELECT
organization_id,
project_id,
environment_id,
task_identifier,
error_fingerprint,
any(coalesce(nullIf(toString(error.data.type), ''), nullIf(toString(error.data.name), ''), 'Error')) as error_type,
any(coalesce(nullIf(substring(toString(error.data.message), 1, 500), ''), 'Unknown error')) as error_message,
any(coalesce(substring(toString(error.data.stack), 1, 2000), '')) as sample_stack_trace,
toDateTime(max(created_at)) as last_seen_date,
min(created_at) as first_seen,
max(created_at) as last_seen,
sumState(toUInt64(1)) as occurrence_count,
uniqState(task_version) as affected_task_versions,
anyState(run_id) as sample_run_id,
anyState(friendly_id) as sample_friendly_id,
sumMapState([status], [toUInt64(1)]) as status_distribution
FROM trigger_dev.task_runs_v2
WHERE
error_fingerprint != ''
AND status IN ('SYSTEM_FAILURE', 'CRASHED', 'INTERRUPTED', 'COMPLETED_WITH_ERRORS', 'TIMED_OUT')
AND _is_deleted = 0
GROUP BY
organization_id,
project_id,
environment_id,
task_identifier,
error_fingerprint;
-- +goose Down
DROP VIEW IF EXISTS trigger_dev.errors_mv_v1;
DROP TABLE IF EXISTS trigger_dev.errors_v1;
@@ -0,0 +1,89 @@
-- +goose Up
-- Per-minute error occurrence counts, keyed by fingerprint + task + version.
-- Powers precise time-range filtering and dynamic-granularity occurrence charts.
CREATE TABLE
trigger_dev.error_occurrences_v1 (
organization_id String,
project_id String,
environment_id String,
task_identifier String,
error_fingerprint String,
task_version String,
minute DateTime,
error_type String,
error_message String,
stack_trace String,
count UInt64,
INDEX idx_error_type_search lower(error_type) TYPE ngrambf_v1 (3, 32768, 2, 0) GRANULARITY 1,
INDEX idx_error_message_search lower(error_message) TYPE ngrambf_v1 (3, 32768, 2, 0) GRANULARITY 1
) ENGINE = SummingMergeTree (count)
PARTITION BY
toDate (minute)
ORDER BY
(
organization_id,
project_id,
environment_id,
task_identifier,
error_fingerprint,
task_version,
minute
) TTL minute + INTERVAL 90 DAY SETTINGS index_granularity = 8192;
CREATE MATERIALIZED VIEW trigger_dev.error_occurrences_mv_v1 TO trigger_dev.error_occurrences_v1 AS
SELECT
organization_id,
project_id,
environment_id,
task_identifier,
error_fingerprint,
task_version,
toStartOfMinute (created_at) as minute,
any (
coalesce(
nullIf(toString (error.data.type), ''),
nullIf(toString (error.data.name), ''),
'Error'
)
) as error_type,
any (
coalesce(
nullIf(
substring(toString (error.data.message), 1, 500),
''
),
'Unknown error'
)
) as error_message,
any (
coalesce(
substring(toString (error.data.stack), 1, 2000),
''
)
) as stack_trace,
count() as count
FROM
trigger_dev.task_runs_v2
WHERE
error_fingerprint != ''
AND status IN (
'SYSTEM_FAILURE',
'CRASHED',
'INTERRUPTED',
'COMPLETED_WITH_ERRORS',
'TIMED_OUT'
)
AND _is_deleted = 0
GROUP BY
organization_id,
project_id,
environment_id,
task_identifier,
error_fingerprint,
task_version,
minute;
-- +goose Down
DROP VIEW IF EXISTS trigger_dev.error_occurrences_mv_v1;
DROP TABLE IF EXISTS trigger_dev.error_occurrences_v1;
@@ -13,6 +13,7 @@ export class ClickhouseQueryBuilder<TOutput> {
private name: string;
private baseQuery: string;
private whereClauses: string[] = [];
private havingClauses: string[] = [];
private params: QueryParams = {};
private orderByClause: string | null = null;
private limitClause: string | null = null;
@@ -69,6 +70,21 @@ export class ClickhouseQueryBuilder<TOutput> {
return this;
}
having(clause: string, params?: QueryParams): this {
this.havingClauses.push(clause);
if (params) {
Object.assign(this.params, params);
}
return this;
}
havingIf(condition: any, clause: string, params?: QueryParams): this {
if (condition) {
this.having(clause, params);
}
return this;
}
orderBy(clause: string): this {
this.orderByClause = clause;
return this;
@@ -101,6 +117,9 @@ export class ClickhouseQueryBuilder<TOutput> {
if (this.groupByClause) {
query += ` GROUP BY ${this.groupByClause}`;
}
if (this.havingClauses.length > 0) {
query += " HAVING " + this.havingClauses.join(" AND ");
}
if (this.orderByClause) {
query += ` ORDER BY ${this.orderByClause}`;
}
@@ -119,6 +138,7 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
private settings: ClickHouseSettings | undefined;
private prewhereClauses: string[] = [];
private whereClauses: string[] = [];
private havingClauses: string[] = [];
private params: QueryParams = {};
private orderByClause: string | null = null;
private limitClause: string | null = null;
@@ -191,6 +211,21 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
return this;
}
having(clause: string, params?: QueryParams): this {
this.havingClauses.push(clause);
if (params) {
Object.assign(this.params, params);
}
return this;
}
havingIf(condition: any, clause: string, params?: QueryParams): this {
if (condition) {
this.having(clause, params);
}
return this;
}
orderBy(clause: string): this {
this.orderByClause = clause;
return this;
@@ -225,6 +260,9 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
if (this.groupByClause) {
query += ` GROUP BY ${this.groupByClause}`;
}
if (this.havingClauses.length > 0) {
query += " HAVING " + this.havingClauses.join(" AND ");
}
if (this.orderByClause) {
query += ` ORDER BY ${this.orderByClause}`;
}
+316
View File
@@ -0,0 +1,316 @@
import { ClickHouseSettings } from "@clickhouse/client";
import { z } from "zod";
import { ClickhouseReader } from "./client/types.js";
import { ClickhouseQueryBuilder } from "./client/queryBuilder.js";
export const ErrorGroupsListQueryResult = z.object({
error_fingerprint: z.string(),
task_identifier: z.string(),
error_type: z.string(),
error_message: z.string(),
first_seen: z.string(),
last_seen: z.string(),
occurrence_count: z.number(),
sample_run_id: z.string(),
sample_friendly_id: z.string(),
});
export type ErrorGroupsListQueryResult = z.infer<typeof ErrorGroupsListQueryResult>;
/**
* Gets a query builder for listing error groups from the pre-aggregated errors_v1 table.
* Allows flexible filtering and pagination.
*/
export function getErrorGroupsListQueryBuilder(
ch: ClickhouseReader,
settings?: ClickHouseSettings
) {
return ch.queryBuilder({
name: "getErrorGroupsList",
baseQuery: `
SELECT
error_fingerprint,
task_identifier,
any(error_type) as error_type,
any(error_message) as error_message,
toString(toUnixTimestamp64Milli(min(first_seen))) as first_seen,
toString(toUnixTimestamp64Milli(max(last_seen))) as last_seen,
toUInt64(sumMerge(occurrence_count)) as occurrence_count,
anyMerge(sample_run_id) as sample_run_id,
anyMerge(sample_friendly_id) as sample_friendly_id
FROM trigger_dev.errors_v1
`,
schema: ErrorGroupsListQueryResult,
settings,
});
}
export const ErrorGroupQueryResult = z.object({
error_fingerprint: z.string(),
task_identifier: z.string(),
error_type: z.string(),
error_message: z.string(),
first_seen: z.string(),
last_seen: z.string(),
occurrence_count: z.number(),
sample_run_id: z.string(),
sample_friendly_id: z.string(),
});
export type ErrorGroupQueryResult = z.infer<typeof ErrorGroupQueryResult>;
export const ErrorGroupQueryParams = z.object({
organizationId: z.string(),
projectId: z.string(),
environmentId: z.string(),
days: z.number().int().default(30),
limit: z.number().int().default(50),
offset: z.number().int().default(0),
});
export type ErrorGroupQueryParams = z.infer<typeof ErrorGroupQueryParams>;
/**
* Gets error groups from the pre-aggregated errors_v1 table.
* Much faster than on-the-fly aggregation.
*/
export function getErrorGroups(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getErrorGroups",
query: `
SELECT
error_fingerprint,
task_identifier,
any(error_type) as error_type,
any(error_message) as error_message,
toString(toUnixTimestamp64Milli(min(first_seen))) as first_seen,
toString(toUnixTimestamp64Milli(max(last_seen))) as last_seen,
toUInt64(sumMerge(occurrence_count)) as occurrence_count,
anyMerge(sample_run_id) as sample_run_id,
anyMerge(sample_friendly_id) as sample_friendly_id
FROM trigger_dev.errors_v1
WHERE
organization_id = {organizationId: String}
AND project_id = {projectId: String}
AND environment_id = {environmentId: String}
GROUP BY error_fingerprint, task_identifier
HAVING max(last_seen) >= now() - INTERVAL {days: Int64} DAY
ORDER BY last_seen DESC
LIMIT {limit: Int64}
OFFSET {offset: Int64}
`,
schema: ErrorGroupQueryResult,
params: ErrorGroupQueryParams,
settings,
});
}
export const ErrorInstanceQueryResult = z.object({
run_id: z.string(),
friendly_id: z.string(),
task_identifier: z.string(),
created_at: z.string(),
status: z.string(),
error_text: z.string(),
trace_id: z.string(),
task_version: z.string(),
});
export type ErrorInstanceQueryResult = z.infer<typeof ErrorInstanceQueryResult>;
export const ErrorInstanceQueryParams = z.object({
organizationId: z.string(),
projectId: z.string(),
environmentId: z.string(),
errorFingerprint: z.string(),
limit: z.number().int().default(50),
offset: z.number().int().default(0),
});
export type ErrorInstanceQueryParams = z.infer<typeof ErrorInstanceQueryParams>;
export const ErrorHourlyOccurrencesQueryResult = z.object({
error_fingerprint: z.string(),
hour_epoch: z.number(),
count: z.number(),
});
export type ErrorHourlyOccurrencesQueryResult = z.infer<typeof ErrorHourlyOccurrencesQueryResult>;
export const ErrorHourlyOccurrencesQueryParams = z.object({
organizationId: z.string(),
projectId: z.string(),
environmentId: z.string(),
fingerprints: z.array(z.string()),
hours: z.number().int().default(24),
});
export type ErrorHourlyOccurrencesQueryParams = z.infer<typeof ErrorHourlyOccurrencesQueryParams>;
/**
* Gets hourly occurrence counts for specific error fingerprints over the past N hours.
* Queries task_runs_v2 directly, grouped by fingerprint and hour.
*/
export function getErrorHourlyOccurrences(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getErrorHourlyOccurrences",
query: `
SELECT
error_fingerprint,
toUnixTimestamp(toStartOfHour(created_at)) as hour_epoch,
count() as count
FROM trigger_dev.task_runs_v2 FINAL
WHERE
organization_id = {organizationId: String}
AND project_id = {projectId: String}
AND environment_id = {environmentId: String}
AND created_at >= now() - INTERVAL {hours: Int64} HOUR
AND error_fingerprint IN {fingerprints: Array(String)}
AND status IN ('SYSTEM_FAILURE', 'CRASHED', 'INTERRUPTED', 'COMPLETED_WITH_ERRORS', 'TIMED_OUT')
AND _is_deleted = 0
GROUP BY
error_fingerprint,
hour_epoch
ORDER BY
error_fingerprint ASC,
hour_epoch ASC
`,
schema: ErrorHourlyOccurrencesQueryResult,
params: ErrorHourlyOccurrencesQueryParams,
settings,
});
}
/**
* Gets individual run instances for a specific error fingerprint.
*/
export function getErrorInstances(ch: ClickhouseReader, settings?: ClickHouseSettings) {
return ch.query({
name: "getErrorInstances",
query: `
SELECT
run_id,
friendly_id,
task_identifier,
toString(created_at) as created_at,
status,
error_text,
trace_id,
task_version
FROM trigger_dev.task_runs_v2 FINAL
WHERE
organization_id = {organizationId: String}
AND project_id = {projectId: String}
AND environment_id = {environmentId: String}
AND error_fingerprint = {errorFingerprint: String}
AND _is_deleted = 0
ORDER BY created_at DESC
LIMIT {limit: Int64}
OFFSET {offset: Int64}
`,
schema: ErrorInstanceQueryResult,
params: ErrorInstanceQueryParams,
settings,
});
}
// ---------------------------------------------------------------------------
// Affected versions distinct task_version from error_occurrences_v1
// ---------------------------------------------------------------------------
export const ErrorAffectedVersionsQueryResult = z.object({
task_version: z.string(),
});
export type ErrorAffectedVersionsQueryResult = z.infer<typeof ErrorAffectedVersionsQueryResult>;
/**
* Query builder for fetching distinct task_version values for an error fingerprint
* from the error_occurrences_v1 SummingMergeTree table.
* task_version is part of the ORDER BY key, so this is efficient.
*/
export function getErrorAffectedVersionsQueryBuilder(
ch: ClickhouseReader,
settings?: ClickHouseSettings
) {
return ch.queryBuilder({
name: "getErrorAffectedVersions",
baseQuery: `
SELECT DISTINCT task_version
FROM trigger_dev.error_occurrences_v1
`,
schema: ErrorAffectedVersionsQueryResult,
settings,
});
}
// ---------------------------------------------------------------------------
// error_occurrences_v1 per-minute bucketed error counts
// ---------------------------------------------------------------------------
export const ErrorOccurrencesListQueryResult = z.object({
error_fingerprint: z.string(),
task_identifier: z.string(),
error_type: z.string(),
error_message: z.string(),
occurrence_count: z.number(),
});
export type ErrorOccurrencesListQueryResult = z.infer<typeof ErrorOccurrencesListQueryResult>;
/**
* Query builder for listing error groups from the per-minute error_occurrences_v1 table.
* Time filtering is done via WHERE on the `minute` column, giving precise time-scoped counts.
*/
export function getErrorOccurrencesListQueryBuilder(
ch: ClickhouseReader,
settings?: ClickHouseSettings
) {
return ch.queryBuilder({
name: "getErrorOccurrencesList",
baseQuery: `
SELECT
error_fingerprint,
task_identifier,
any(error_type) as error_type,
any(error_message) as error_message,
sum(count) as occurrence_count
FROM trigger_dev.error_occurrences_v1
`,
schema: ErrorOccurrencesListQueryResult,
settings,
});
}
export const ErrorOccurrencesBucketQueryResult = z.object({
error_fingerprint: z.string(),
bucket_epoch: z.number(),
count: z.number(),
});
export type ErrorOccurrencesBucketQueryResult = z.infer<typeof ErrorOccurrencesBucketQueryResult>;
/**
* Creates a query builder for bucketed error occurrence counts.
* The `intervalExpr` is a ClickHouse INTERVAL literal (e.g. "INTERVAL 1 HOUR").
* Returns a builder directly since the base query varies with each granularity.
*/
export function createErrorOccurrencesQueryBuilder(
ch: ClickhouseReader,
intervalExpr: string,
settings?: ClickHouseSettings
): ClickhouseQueryBuilder<ErrorOccurrencesBucketQueryResult> {
return new ClickhouseQueryBuilder(
"getErrorOccurrencesBucketed",
`
SELECT
error_fingerprint,
toUnixTimestamp(toStartOfInterval(minute, ${intervalExpr})) as bucket_epoch,
sum(count) as count
FROM trigger_dev.error_occurrences_v1
`,
ch,
ErrorOccurrencesBucketQueryResult,
settings
);
}
+24
View File
@@ -27,6 +27,16 @@ import {
getLogsSearchListQueryBuilder,
} from "./taskEvents.js";
import { insertMetrics } from "./metrics.js";
import {
getErrorGroups,
getErrorInstances,
getErrorGroupsListQueryBuilder,
getErrorHourlyOccurrences,
getErrorOccurrencesListQueryBuilder,
createErrorOccurrencesQueryBuilder,
getErrorAffectedVersionsQueryBuilder,
} from "./errors.js";
export { msToClickHouseInterval } from "./intervals.js";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
@@ -34,6 +44,7 @@ import type { Agent as HttpsAgent } from "https";
export type * from "./taskRuns.js";
export type * from "./taskEvents.js";
export type * from "./metrics.js";
export type * from "./errors.js";
export type * from "./client/queryBuilder.js";
// Re-export column constants, indices, and type-safe accessors
@@ -229,4 +240,17 @@ export class ClickHouse {
logsListQueryBuilder: getLogsSearchListQueryBuilder(this.reader),
};
}
get errors() {
return {
getGroups: getErrorGroups(this.reader),
getInstances: getErrorInstances(this.reader),
getHourlyOccurrences: getErrorHourlyOccurrences(this.reader),
affectedVersionsQueryBuilder: getErrorAffectedVersionsQueryBuilder(this.reader),
listQueryBuilder: getErrorGroupsListQueryBuilder(this.reader),
occurrencesListQueryBuilder: getErrorOccurrencesListQueryBuilder(this.reader),
createOccurrencesQueryBuilder: (intervalExpr: string) =>
createErrorOccurrencesQueryBuilder(this.reader, intervalExpr),
};
}
}
@@ -0,0 +1,5 @@
/** Converts a granularity in milliseconds to a ClickHouse INTERVAL expression. */
export function msToClickHouseInterval(ms: number): string {
const seconds = Math.round(ms / 1000);
return `INTERVAL ${seconds} SECOND`;
}
@@ -52,7 +52,15 @@ describe("Task Runs V2", () => {
100, // cost_in_cents
0, // base_cost_in_cents
{ data: { key: "value" } }, // output
{ data: { type: "BUILT_IN_ERROR", name: "Error", message: "error", stackTrace: "stack trace" } }, // error
{
data: {
type: "BUILT_IN_ERROR",
name: "Error",
message: "error",
stackTrace: "stack trace",
},
}, // error
"1234567890", // error_fingerprint
["tag1", "tag2"], // tags
"1.0.0", // task_version
"1.0.0", // sdk_version
@@ -180,6 +188,7 @@ describe("Task Runs V2", () => {
0, // base_cost_in_cents
{ data: null }, // output
{ data: null }, // error
"", // error_fingerprint
[], // tags
"", // task_version
"", // sdk_version
@@ -230,6 +239,7 @@ describe("Task Runs V2", () => {
0, // base_cost_in_cents
{ data: null }, // output
{ data: null }, // error
"", // error_fingerprint
[], // tags
"", // task_version
"", // sdk_version
@@ -327,6 +337,7 @@ describe("Task Runs V2", () => {
0, // base_cost_in_cents
{ data: null }, // output
{ data: null }, // error
"", // error_fingerprint
[], // tags
"", // task_version
"", // sdk_version
@@ -29,6 +29,7 @@ export const TaskRunV2 = z.object({
base_cost_in_cents: z.number().default(0),
output: z.unknown(),
error: z.unknown(),
error_fingerprint: z.string().default(""),
tags: z.array(z.string()).default([]),
task_version: z.string(),
sdk_version: z.string(),
@@ -82,6 +83,7 @@ export const TASK_RUN_COLUMNS = [
"base_cost_in_cents",
"output",
"error",
"error_fingerprint",
"tags",
"task_version",
"sdk_version",
@@ -144,6 +146,7 @@ export type TaskRunFieldTypes = {
base_cost_in_cents: number;
output: { data: unknown };
error: { data: unknown };
error_fingerprint: string;
tags: string[];
task_version: string;
sdk_version: string;
@@ -277,6 +280,7 @@ export type TaskRunInsertArray = [
base_cost_in_cents: number,
output: { data: unknown },
error: { data: unknown },
error_fingerprint: string,
tags: string[],
task_version: string,
sdk_version: string,
@@ -96,6 +96,7 @@ export const WaitpointId = new IdUtil("waitpoint");
export const BatchId = new IdUtil("batch");
export const BulkActionId = new IdUtil("bulk");
export const AttemptId = new IdUtil("attempt");
export const ErrorId = new IdUtil("error");
export class IdGenerator {
private alphabet: string;