Create new partitioned TaskEvent table, and switch to it gradually as new runs are created (#1696)

* Create new partitioned TaskEvent table, and switch to it gradually as new runs are created

* Add env var for partition window in seconds

* Make startCreatedAt required in task event store

Author: ericallam

apps/webapp/app/components/runs/v3/RunInspector.tsx

@@ -32,14 +32,11 @@ import { cn } from "~/utils/cn";
 import { formatCurrencyAccurate } from "~/utils/numberFormatter";
 import {
   v3RunDownloadLogsPath,
-  v3RunPath,
   v3RunSpanPath,
   v3RunsPath,
   v3SchedulePath,
-  v3TraceSpanPath,
 } from "~/utils/pathBuilder";
 import { TraceSpan } from "~/utils/taskEvent";
-import { SpanLink } from "~/v3/eventRepository.server";
 import { isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
 import { RunTimelineEvent, RunTimelineLine } from "./InspectorTimeline";
 import { RunTag } from "./RunTag";
@@ -317,18 +314,6 @@ export function RunInspector({
                     )}
                   </Property.Value>
                 </Property.Item>
-                {span?.links && span.links.length > 0 && (
-                  <Property.Item>
-                    <Property.Label>Links</Property.Label>
-                    <Property.Value>
-                      <div className="space-y-1">
-                        {span.links.map((link, index) => (
-                          <SpanLinkElement key={index} link={link} />
-                        ))}
-                      </div>
-                    </Property.Value>
-                  </Property.Item>
-                )}
                 <Property.Item>
                   <Property.Label>Max duration</Property.Label>
                   <Property.Value>
@@ -647,27 +632,3 @@ function PacketDisplay({
     }
   }
 }
-
-function SpanLinkElement({ link }: { link: SpanLink }) {
-  const organization = useOrganization();
-  const project = useProject();
-
-  switch (link.type) {
-    case "run": {
-      return (
-        <TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
-          {link.title}
-        </TextLink>
-      );
-    }
-    case "span": {
-      return (
-        <TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
-          {link.title}
-        </TextLink>
-      );
-    }
-  }
-
-  return null;
-}

apps/webapp/app/components/runs/v3/SpanInspector.tsx

@@ -1,9 +1,11 @@
+import { formatDuration, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { CodeBlock } from "~/components/code/CodeBlock";
 import { Button } from "~/components/primitives/Buttons";
 import { DateTimeAccurate } from "~/components/primitives/DateTime";
 import { Header2 } from "~/components/primitives/Headers";
 import * as Property from "~/components/primitives/PropertyTable";
+import { Spinner } from "~/components/primitives/Spinner";
 import { TabButton, TabContainer } from "~/components/primitives/Tabs";
 import { TextLink } from "~/components/primitives/TextLink";
 import { InfoIconTooltip, SimpleTooltip } from "~/components/primitives/Tooltip";
@@ -15,13 +17,10 @@ import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { cn } from "~/utils/cn";
-import { v3RunPath, v3RunsPath, v3TraceSpanPath } from "~/utils/pathBuilder";
+import { v3RunsPath } from "~/utils/pathBuilder";
 import { TraceSpan } from "~/utils/taskEvent";
-import { SpanLink } from "~/v3/eventRepository.server";
 import { RunTimelineEvent, RunTimelineLine } from "./InspectorTimeline";
-import { Spinner } from "~/components/primitives/Spinner";
 import { LiveTimer } from "./LiveTimer";
-import { formatDuration, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
 
 export function SpanInspector({
   span,
@@ -150,18 +149,6 @@ export function SpanInspector({
                     )}
                   </Property.Value>
                 </Property.Item>
-                {span.links && span.links.length > 0 && (
-                  <Property.Item>
-                    <Property.Label>Links</Property.Label>
-                    <Property.Value>
-                      <div className="space-y-1">
-                        {span.links.map((link, index) => (
-                          <SpanLinkElement key={index} link={link} />
-                        ))}
-                      </div>
-                    </Property.Value>
-                  </Property.Item>
-                )}
               </Property.Table>
             </div>
           ) : (
@@ -203,18 +190,6 @@ export function SpanInspector({
                   <Property.Label>Message</Property.Label>
                   <Property.Value>{span.message}</Property.Value>
                 </Property.Item>
-                {span.links && span.links.length > 0 && (
-                  <Property.Item>
-                    <Property.Label>Links</Property.Label>
-                    <Property.Value>
-                      <div className="space-y-1">
-                        {span.links.map((link, index) => (
-                          <SpanLinkElement key={index} link={link} />
-                        ))}
-                      </div>
-                    </Property.Value>
-                  </Property.Item>
-                )}
               </Property.Table>
 
               {span.events !== undefined && <SpanEvents spanEvents={span.events} />}
@@ -287,27 +262,3 @@ export function SpanTimeline({ startTime, duration, inProgress, isError }: Timel
     </>
   );
 }
-
-function SpanLinkElement({ link }: { link: SpanLink }) {
-  const organization = useOrganization();
-  const project = useProject();
-
-  switch (link.type) {
-    case "run": {
-      return (
-        <TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
-          {link.title}
-        </TextLink>
-      );
-    }
-    case "span": {
-      return (
-        <TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
-          {link.title}
-        </TextLink>
-      );
-    }
-  }
-
-  return null;
-}

apps/webapp/app/env.server.ts

@@ -481,6 +481,9 @@ const EnvironmentSchema = z.object({
     .transform((v) => v ?? process.env.REDIS_PASSWORD),
   COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
   COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+
+  TASK_EVENT_PARTITIONING_ENABLED: z.string().default("0"),
+  TASK_EVENT_PARTITIONED_WINDOW_IN_SECONDS: z.coerce.number().int().default(60), // 1 minute
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/presenters/v3/RunPresenter.server.ts

@@ -3,6 +3,7 @@ import { createTreeFromFlatItems, flattenTree } from "~/components/primitives/Tr
 import { PrismaClient, prisma } from "~/db.server";
 import { getUsername } from "~/utils/username";
 import { eventRepository } from "~/v3/eventRepository.server";
+import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 import { isFinalRunStatus } from "~/v3/taskStatus";
 
 type Result = Awaited<ReturnType<RunPresenter["call"]>>;
@@ -32,6 +33,8 @@ export class RunPresenter {
     const run = await this.#prismaClient.taskRun.findFirstOrThrow({
       select: {
         id: true,
+        createdAt: true,
+        taskEventStore: true,
         number: true,
         traceId: true,
         spanId: true,
@@ -105,7 +108,12 @@ export class RunPresenter {
     }
 
     // get the events
-    const traceSummary = await eventRepository.getTraceSummary(run.traceId);
+    const traceSummary = await eventRepository.getTraceSummary(
+      getTaskEventStoreTableForRun(run),
+      run.traceId,
+      run.createdAt,
+      run.completedAt ?? undefined
+    );
     if (!traceSummary) {
       return {
         run: runData,

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -10,6 +10,7 @@ import { machinePresetFromName } from "~/v3/machinePresets.server";
 import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
 import { BasePresenter } from "./basePresenter.server";
 import { getMaxDuration } from "~/v3/utils/maxDuration";
+import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 
 type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
 export type Span = NonNullable<NonNullable<Result>["span"]>;
@@ -71,6 +72,7 @@ export class SpanPresenter extends BasePresenter {
         friendlyId: true,
         isTest: true,
         maxDurationInSeconds: true,
+        taskEventStore: true,
         tags: {
           select: {
             name: true,
@@ -205,7 +207,13 @@ export class SpanPresenter extends BasePresenter {
       }
     }
 
-    const span = await eventRepository.getSpan(spanId, run.traceId);
+    const span = await eventRepository.getSpan(
+      getTaskEventStoreTableForRun(run),
+      spanId,
+      run.traceId,
+      run.createdAt,
+      run.completedAt ?? undefined
+    );
 
     const metadata = run.metadata
       ? await prettyPrintPacket(run.metadata, run.metadataType, {
@@ -342,6 +350,9 @@ export class SpanPresenter extends BasePresenter {
     const run = await this._prisma.taskRun.findFirst({
       select: {
         traceId: true,
+        createdAt: true,
+        completedAt: true,
+        taskEventStore: true,
       },
       where: {
         friendlyId: runFriendlyId,
@@ -352,7 +363,13 @@ export class SpanPresenter extends BasePresenter {
       return;
     }
 
-    const span = await eventRepository.getSpan(spanId, run.traceId);
+    const span = await eventRepository.getSpan(
+      getTaskEventStoreTableForRun(run),
+      spanId,
+      run.traceId,
+      run.createdAt,
+      run.completedAt ?? undefined
+    );
 
     if (!span) {
       return;

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.traces.$traceId.spans.$spanId/route.tsx

@@ -44,7 +44,6 @@ import { RunTag } from "~/components/runs/v3/RunTag";
 import { SpanEvents } from "~/components/runs/v3/SpanEvents";
 import { SpanTitle } from "~/components/runs/v3/SpanTitle";
 import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptStatus";
-import { TaskRunsTable } from "~/components/runs/v3/TaskRunsTable";
 import { TaskRunStatusCombo } from "~/components/runs/v3/TaskRunStatus";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
@@ -57,16 +56,13 @@ import { cn } from "~/utils/cn";
 import { formatCurrencyAccurate } from "~/utils/numberFormatter";
 import {
   v3BatchPath,
-  v3BatchRunsPath,
   v3RunDownloadLogsPath,
   v3RunPath,
   v3RunSpanPath,
   v3RunsPath,
   v3SchedulePath,
   v3SpanParamsSchema,
-  v3TraceSpanPath,
 } from "~/utils/pathBuilder";
-import { SpanLink } from "~/v3/eventRepository.server";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
@@ -1152,27 +1148,3 @@ function classNameForState(state: TimelineState) {
     }
   }
 }
-
-function SpanLinkElement({ link }: { link: SpanLink }) {
-  const organization = useOrganization();
-  const project = useProject();
-
-  switch (link.type) {
-    case "run": {
-      return (
-        <TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
-          {link.title}
-        </TextLink>
-      );
-    }
-    case "span": {
-      return (
-        <TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
-          {link.title}
-        </TextLink>
-      );
-    }
-  }
-
-  return null;
-}

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -7,6 +7,7 @@ import { createGzip } from "zlib";
 import { Readable } from "stream";
 import { formatDurationMilliseconds } from "@trigger.dev/core/v3/utils/durations";
 import { getDateFromNanoseconds } from "~/utils/taskEvent";
+import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   const userId = await requireUserId(request);
@@ -31,7 +32,12 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
-  const runEvents = await eventRepository.getRunEvents(run.friendlyId);
+  const runEvents = await eventRepository.getRunEvents(
+    getTaskEventStoreTableForRun(run),
+    run.friendlyId,
+    run.createdAt,
+    run.completedAt ?? undefined
+  );
 
   // Create a Readable stream from the runEvents array
   const readable = new Readable({

apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts

@@ -396,15 +396,6 @@ export function v3RunSpanPath(
   return `${v3RunPath(organization, project, run)}?span=${span.spanId}`;
 }
 
-export function v3TraceSpanPath(
-  organization: OrgForPath,
-  project: ProjectForPath,
-  traceId: string,
-  spanId: string
-) {
-  return `${v3ProjectPath(organization, project)}/traces/${traceId}/spans/${spanId}`;
-}
-
 export function v3RunStreamingPath(
   organization: OrgForPath,
   project: ProjectForPath,

apps/webapp/app/utils/pathBuilder.ts

@@ -62,7 +62,6 @@ export function prepareTrace(events: TaskEvent[]): TraceSummary | undefined {
     );
 
     const span = {
-      recordId: event.id,
       id: event.spanId,
       parentId: event.parentId ?? undefined,
       runId: event.runId,
@@ -247,7 +246,6 @@ export function createSpanFromEvent(events: TaskEvent[], event: PreparedEvent) {
   }
 
   const span = {
-    recordId: event.id,
     id: event.spanId,
     parentId: event.parentId ?? undefined,
     runId: event.runId,