fix(core): stop custom metric exporters breaking the metrics export (#4613)

## Summary

Projects that configure their own `metricExporters` or `metricReaders`
in `trigger.config.ts` were losing task metrics on nearly every run, and
seeing an unexplained `Failed to flush tracingSDK` alongside
`OTLPExporterError: Bad Request` in their run logs. Spans and logs kept
working, so the runs otherwise looked healthy.

## Root cause and fix

Every configured exporter gets its own `PeriodicExportingMetricReader`,
and `meterProvider.forceFlush()` fans out across all readers with
`Promise.all`, so two collections can land on the same millisecond.
`@opentelemetry/host-metrics` divides by the elapsed interval to compute
`process.cpu.utilization`
([common.ts](https://github.com/open-telemetry/opentelemetry-js-contrib/blob/main/packages/host-metrics/src/stats/common.ts)),
so a zero interval yields `0/0`. `JSON.stringify(NaN)` is `null`, and a
collector rejects `"asDouble": null` with a 400 that drops the
**entire** request, not just the offending point.

`flush()` and `shutdown()` now walk the metric readers one at a time, so
collections can no longer share a timestamp. Each reader is isolated, so
one failing reader cannot skip the readers behind it, and every failure
is logged with the reader that produced it. The first error is still
rethrown, so callers see failures exactly as before.

As a second layer, non-finite data points are dropped just before our
own export, so a metric that divides by zero cannot take the rest of the
batch with it. Exporters and readers supplied through
`trigger.config.ts` are untouched by that filter and still receive raw
data.

The trade-off is that configured exporters now flush after the built-in
one rather than alongside it, so flush latency is the sum rather than
the max.

An internal test package's dependency on core was replaced with a local
helper, because core now needs that package in `devDependencies` and the
two together formed a workspace cycle.

## Verification

Tested against a real collector in a container: a batch containing a
`NaN` reading is rejected with a 400 without the fix and accepted with
it, and a single flush is asserted to collect from one reader at a time.

.changeset/lucky-pillows-invite.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Task metrics no longer go missing for projects that configure their own `metricExporters` or `metricReaders`, and the flush error that came with it is gone.

.github/workflows/unit-tests-packages.yml

@@ -99,6 +99,7 @@ jobs:
           pull redis:7.2
           pull testcontainers/ryuk:0.14.0
           pull electricsql/electric:1.2.4
+          pull otel/opentelemetry-collector-k8s:0.158.0@sha256:c09130a633196a5becee164411473a0932ecf223f94fda6dab5f22798ff9f376
           echo "Image pre-pull complete"
 
       - name: 📥 Download deps

internal-packages/testcontainers/package.json

@@ -22,7 +22,6 @@
     "@internal/run-ops-database": "workspace:*",
     "@testcontainers/postgresql": "^11.14.0",
     "@testcontainers/redis": "^11.14.0",
-    "@trigger.dev/core": "workspace:*",
     "std-env": "^3.9.0",
     "testcontainers": "^11.14.0",
     "tinyexec": "^0.3.0"

internal-packages/testcontainers/src/index.ts

@@ -29,6 +29,7 @@ import {
 } from "./utils";
 
 export { assertNonNullable, createPostgresContainer } from "./utils";
+export { OtelCollectorContainer, StartedOtelCollectorContainer } from "./otelCollector";
 export { laggingReplica, type LaggingModel } from "./laggingReplica";
 export { logCleanup };
 export type { MinIOConnectionConfig };

internal-packages/testcontainers/src/otelCollector.ts

@@ -0,0 +1,63 @@
+import type { StartedTestContainer } from "testcontainers";
+import { AbstractStartedContainer, GenericContainer, Wait } from "testcontainers";
+
+const OTLP_HTTP_PORT = 4318;
+const CONFIG_PATH = "/etc/otelcol-config.yaml";
+
+const CONFIG = `receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: 0.0.0.0:${OTLP_HTTP_PORT}
+exporters:
+  debug: {}
+service:
+  telemetry:
+    logs:
+      level: WARN
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [debug]
+    metrics:
+      receivers: [otlp]
+      exporters: [debug]
+    logs:
+      receivers: [otlp]
+      exporters: [debug]
+`;
+
+export class OtelCollectorContainer extends GenericContainer {
+  constructor(
+    image = "otel/opentelemetry-collector-k8s:0.158.0@sha256:c09130a633196a5becee164411473a0932ecf223f94fda6dab5f22798ff9f376"
+  ) {
+    super(image);
+    this.withExposedPorts(OTLP_HTTP_PORT);
+    this.withCopyContentToContainer([{ content: CONFIG, target: CONFIG_PATH }]);
+    this.withCommand([`--config=${CONFIG_PATH}`]);
+    this.withWaitStrategy(Wait.forHttp("/v1/metrics", OTLP_HTTP_PORT).forStatusCode(405));
+    this.withStartupTimeout(120_000);
+  }
+
+  public override async start(): Promise<StartedOtelCollectorContainer> {
+    return new StartedOtelCollectorContainer(await super.start());
+  }
+}
+
+export class StartedOtelCollectorContainer extends AbstractStartedContainer {
+  constructor(startedTestContainer: StartedTestContainer) {
+    super(startedTestContainer);
+  }
+
+  public getPort(): number {
+    return super.getMappedPort(OTLP_HTTP_PORT);
+  }
+
+  /**
+   * Base URL for OTLP/HTTP, without a signal path.
+   * Example: `http://localhost:32768`
+   */
+  public getOtlpHttpUrl(): string {
+    return `http://${this.getHost()}:${this.getPort()}`;
+  }
+}

internal-packages/testcontainers/src/utils.ts

@@ -4,7 +4,6 @@ import { PostgreSqlContainer } from "@testcontainers/postgresql";
 import type { StartedRedisContainer } from "@testcontainers/redis";
 import { RedisContainer } from "@testcontainers/redis";
 import { PrismaClient } from "@trigger.dev/database";
-import { tryCatch } from "@trigger.dev/core";
 import Redis from "ioredis";
 import path from "path";
 import { isDebug } from "std-env";
@@ -16,6 +15,14 @@ import { ClickHouseContainer, runClickhouseMigrations } from "./clickhouse";
 import { MinIOContainer } from "./minio";
 import { getContainerMetadata, getTaskMetadata, logCleanup, logSetup } from "./logs";
 
+async function tryCatch<T, E = Error>(promise: Promise<T>): Promise<[E, null] | [null, T]> {
+  try {
+    return [null, await promise];
+  } catch (error) {
+    return [error as E, null];
+  }
+}
+
 /** Returns the container's connection URI with the database path swapped to `database`. */
 export function postgresUriWithDatabase(uri: string, database: string): string {
   const url = new URL(uri);

packages/core/package.json

@@ -233,6 +233,7 @@
     "@ai-sdk/provider-utils": "^1.0.22",
     "@arethetypeswrong/cli": "^0.18.5",
     "@epic-web/test-server": "^0.1.0",
+    "@internal/testcontainers": "workspace:*",
     "@trigger.dev/database": "workspace:*",
     "@types/humanize-duration": "^3.27.1",
     "@types/lodash.get": "^4.4.9",

packages/core/src/v3/otel/tracingSDK.test.ts

@@ -0,0 +1,234 @@
+import {
+  OtelCollectorContainer,
+  type StartedOtelCollectorContainer,
+} from "@internal/testcontainers";
+
+import { metrics } from "@opentelemetry/api";
+import { ExportResultCode } from "@opentelemetry/core";
+import {
+  MetricReader,
+  type PushMetricExporter,
+  type ResourceMetrics,
+} from "@opentelemetry/sdk-metrics";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { TracingSDK } from "./tracingSDK.js";
+
+class NoopMetricExporter implements PushMetricExporter {
+  forceFlushCount = 0;
+
+  export(_metrics: ResourceMetrics, resultCallback: (result: { code: number }) => void): void {
+    resultCallback({ code: ExportResultCode.SUCCESS });
+  }
+
+  async forceFlush(): Promise<void> {
+    this.forceFlushCount++;
+  }
+
+  async shutdown(): Promise<void> {}
+}
+
+describe("TracingSDK with an external metric exporter", () => {
+  let collector: StartedOtelCollectorContainer;
+  let tracingSDK: TracingSDK;
+
+  beforeAll(async () => {
+    collector = await new OtelCollectorContainer().start();
+
+    process.env.TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS = "600000";
+
+    tracingSDK = new TracingSDK({
+      url: collector.getOtlpHttpUrl(),
+      forceFlushTimeoutMillis: 30_000,
+      diagLogLevel: "none",
+      metricExporters: [new NoopMetricExporter()],
+      hostMetrics: true,
+      hostMetricGroups: ["process.cpu", "process.memory"],
+      nodejsRuntimeMetrics: true,
+    });
+  }, 180_000);
+
+  afterAll(async () => {
+    await tracingSDK?.shutdown();
+    await collector?.stop();
+    delete process.env.TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS;
+  });
+
+  it("flushes without the collector rejecting a batch containing a NaN reading", async () => {
+    const gauge = metrics.getMeter("test").createObservableGauge("test.utilization");
+    gauge.addCallback((result) => result.observe(NaN));
+
+    await expect(tracingSDK.flush()).resolves.toBeUndefined();
+  });
+
+  it("collects from each metric reader one at a time", async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+
+    const gauge = metrics.getMeter("test").createObservableGauge("test.concurrency");
+    gauge.addCallback(async (result) => {
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      result.observe(1);
+      inFlight--;
+    });
+
+    await tracingSDK.flush();
+
+    expect(maxInFlight).toBe(1);
+  });
+});
+
+class FailingMetricReader extends MetricReader {
+  protected async onForceFlush(): Promise<void> {
+    throw new Error("reader flush failed");
+  }
+
+  protected async onShutdown(): Promise<void> {}
+}
+
+class FailingShutdownMetricReader extends MetricReader {
+  shutdownAttempts = 0;
+
+  protected async onForceFlush(): Promise<void> {}
+
+  protected async onShutdown(): Promise<void> {
+    this.shutdownAttempts++;
+    throw new Error(`reader shutdown failed (attempt ${this.shutdownAttempts})`);
+  }
+}
+
+class RecordingMetricReader extends MetricReader {
+  forceFlushCount = 0;
+  shutdownCount = 0;
+
+  protected async onForceFlush(): Promise<void> {
+    this.forceFlushCount++;
+  }
+
+  protected async onShutdown(): Promise<void> {
+    this.shutdownCount++;
+  }
+}
+
+function captureConsoleErrors(): { lines: string[]; restore: () => void } {
+  const lines: string[] = [];
+  const original = console.error;
+
+  console.error = (...args: unknown[]) => {
+    lines.push(args.map(String).join(" "));
+  };
+
+  return { lines, restore: () => (console.error = original) };
+}
+
+describe("TracingSDK when one metric reader fails to flush", () => {
+  let recordingReader: RecordingMetricReader;
+  let tracingSDK: TracingSDK;
+
+  beforeAll(() => {
+    recordingReader = new RecordingMetricReader();
+
+    tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingMetricReader(), recordingReader],
+    });
+  });
+
+  it("still flushes the readers after it", async () => {
+    await tracingSDK.flush().catch(() => {});
+
+    expect(recordingReader.forceFlushCount).toBeGreaterThan(0);
+  });
+
+  it("still reports the failure to the caller", async () => {
+    await expect(tracingSDK.flush()).rejects.toThrow("reader flush failed");
+  });
+
+  it("logs the failure as it happens", async () => {
+    const console = captureConsoleErrors();
+
+    await tracingSDK.flush().catch(() => {});
+    console.restore();
+
+    expect(console.lines.join("\n")).toContain("reader flush failed");
+  });
+});
+
+class OverlapRecordingMetricReader extends MetricReader {
+  static inFlight = 0;
+  static maxInFlight = 0;
+
+  protected async onForceFlush(): Promise<void> {}
+
+  protected async onShutdown(): Promise<void> {
+    OverlapRecordingMetricReader.inFlight++;
+    OverlapRecordingMetricReader.maxInFlight = Math.max(
+      OverlapRecordingMetricReader.maxInFlight,
+      OverlapRecordingMetricReader.inFlight
+    );
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    OverlapRecordingMetricReader.inFlight--;
+  }
+}
+
+describe("TracingSDK shutdown", () => {
+  it("shuts down each metric reader one at a time", async () => {
+    OverlapRecordingMetricReader.inFlight = 0;
+    OverlapRecordingMetricReader.maxInFlight = 0;
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new OverlapRecordingMetricReader(), new OverlapRecordingMetricReader()],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(OverlapRecordingMetricReader.maxInFlight).toBe(1);
+  });
+
+  it("still shuts down the readers after one that fails", async () => {
+    const recordingReader = new RecordingMetricReader();
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingShutdownMetricReader(), recordingReader],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(recordingReader.shutdownCount).toBeGreaterThan(0);
+  });
+
+  it("does not retry a metric reader that failed to shut down", async () => {
+    const failingReader = new FailingShutdownMetricReader();
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [failingReader],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(failingReader.shutdownAttempts).toBe(1);
+  });
+
+  it("reports the original shutdown failure, not a later one", async () => {
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingShutdownMetricReader()],
+    });
+
+    await expect(tracingSDK.shutdown()).rejects.toThrow("attempt 1");
+  });
+});

packages/core/src/v3/otel/tracingSDK.ts

@@ -100,6 +100,7 @@ export class TracingSDK {
   private readonly _spanExporter: SpanExporter;
   private readonly _traceProvider: NodeTracerProvider;
   private readonly _meterProvider: MeterProvider;
+  private readonly _metricReaders: MetricReader[];
 
   public readonly getLogger: LoggerProvider["getLogger"];
   public readonly getTracer: TracerProvider["getTracer"];
@@ -318,6 +319,7 @@ export class TracingSDK {
     });
 
     this._meterProvider = meterProvider;
+    this._metricReaders = metricReaders;
     metrics.setGlobalMeterProvider(meterProvider);
 
     if (config.hostMetrics) {
@@ -348,15 +350,44 @@ export class TracingSDK {
     await Promise.all([
       this._traceProvider.forceFlush(),
       this._logProvider.forceFlush(),
-      this._meterProvider.forceFlush(),
+      this._flushMetricReadersSerially(),
     ]);
   }
 
+  private async _flushMetricReadersSerially() {
+    await this._eachMetricReaderSerially("flush", (reader) => reader.forceFlush());
+  }
+
+  private async _shutdownMetricReadersSerially() {
+    await this._eachMetricReaderSerially("shut down", (reader) => reader.shutdown());
+    await this._meterProvider.shutdown();
+  }
+
+  private async _eachMetricReaderSerially(
+    action: string,
+    run: (reader: MetricReader) => Promise<void>
+  ) {
+    const errors: unknown[] = [];
+
+    for (const reader of this._metricReaders) {
+      try {
+        await run(reader);
+      } catch (error) {
+        console.error(`Failed to ${action} metric reader ${reader.constructor.name}`, error);
+        errors.push(error);
+      }
+    }
+
+    if (errors.length > 0) {
+      throw errors[0];
+    }
+  }
+
   public async shutdown() {
     await Promise.all([
       this._traceProvider.shutdown(),
       this._logProvider.shutdown(),
-      this._meterProvider.shutdown(),
+      this._shutdownMetricReadersSerially(),
     ]);
   }
 }

packages/core/src/v3/otel/tracingSDKShutdown.test.ts

@@ -0,0 +1,19 @@
+import { metrics } from "@opentelemetry/api";
+import { describe, expect, it } from "vitest";
+import { TracingSDK } from "./tracingSDK.js";
+
+describe("TracingSDK shutdown", () => {
+  it("leaves the meter provider shut down", async () => {
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+    });
+
+    const meterBeforeShutdown = metrics.getMeter("test");
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(metrics.getMeter("test")).not.toBe(meterBeforeShutdown);
+  });
+});

packages/core/src/v3/taskContext/otelProcessors.test.ts

@@ -0,0 +1,144 @@
+import {
+  OtelCollectorContainer,
+  type StartedOtelCollectorContainer,
+} from "@internal/testcontainers";
+import { ExportResultCode } from "@opentelemetry/core";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
+import {
+  AggregationTemporality,
+  DataPointType,
+  InstrumentType,
+  type MetricData,
+  type PushMetricExporter,
+  type ResourceMetrics,
+} from "@opentelemetry/sdk-metrics";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { BufferingMetricExporter } from "./otelProcessors.js";
+
+class RecordingMetricExporter implements PushMetricExporter {
+  exported: ResourceMetrics[] = [];
+
+  export(metrics: ResourceMetrics, resultCallback: (result: { code: number }) => void): void {
+    this.exported.push(metrics);
+    resultCallback({ code: ExportResultCode.SUCCESS });
+  }
+
+  async forceFlush(): Promise<void> {}
+  async shutdown(): Promise<void> {}
+}
+
+function gauge(name: string, values: number[]): MetricData {
+  return {
+    descriptor: {
+      name,
+      description: "",
+      unit: "1",
+      type: InstrumentType.OBSERVABLE_GAUGE,
+      valueType: 1,
+    },
+    aggregationTemporality: AggregationTemporality.CUMULATIVE,
+    dataPointType: DataPointType.GAUGE,
+    dataPoints: values.map((value) => ({
+      attributes: { "process.cpu.state": "user" },
+      startTime: [1786481102, 584000000],
+      endTime: [1786481102, 698000000],
+      value,
+    })),
+  } as MetricData;
+}
+
+function resourceMetrics(metrics: MetricData[]): ResourceMetrics {
+  return {
+    resource: { attributes: {} },
+    scopeMetrics: [{ scope: { name: "@opentelemetry/host-metrics" }, metrics }],
+  } as unknown as ResourceMetrics;
+}
+
+async function exportThrough(exporter: BufferingMetricExporter, metrics: ResourceMetrics) {
+  exporter.export(metrics, () => {});
+  await exporter.forceFlush();
+}
+
+function histogram(name: string, sums: number[]): MetricData {
+  return {
+    descriptor: {
+      name,
+      description: "",
+      unit: "ms",
+      type: InstrumentType.HISTOGRAM,
+      valueType: 1,
+    },
+    aggregationTemporality: AggregationTemporality.DELTA,
+    dataPointType: DataPointType.HISTOGRAM,
+    dataPoints: sums.map((sum) => ({
+      attributes: { "task.status": "completed" },
+      startTime: [1786481102, 584000000],
+      endTime: [1786481102, 698000000],
+      value: {
+        min: 0,
+        max: 1,
+        sum,
+        count: 1,
+        buckets: { boundaries: [0, 1], counts: [0, 1, 0] },
+      },
+    })),
+  } as MetricData;
+}
+
+describe("BufferingMetricExporter", () => {
+  it("forwards only the finite data points of a metric", async () => {
+    const inner = new RecordingMetricExporter();
+    const exporter = new BufferingMetricExporter(inner, 30_000);
+
+    await exportThrough(exporter, resourceMetrics([gauge("process.cpu.utilization", [NaN, 0.25])]));
+
+    const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!;
+    expect(forwarded.dataPoints.map((dp) => dp.value)).toEqual([0.25]);
+  });
+
+  it("drops infinite data points as well as NaN", async () => {
+    const inner = new RecordingMetricExporter();
+    const exporter = new BufferingMetricExporter(inner, 30_000);
+
+    await exportThrough(
+      exporter,
+      resourceMetrics([gauge("process.cpu.utilization", [Infinity, -Infinity, 0.5])])
+    );
+
+    const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!;
+    expect(forwarded.dataPoints.map((dp) => dp.value)).toEqual([0.5]);
+  });
+
+  it("forwards only the histogram data points with a finite sum", async () => {
+    const inner = new RecordingMetricExporter();
+    const exporter = new BufferingMetricExporter(inner, 30_000);
+
+    await exportThrough(exporter, resourceMetrics([histogram("task.duration", [NaN, 5])]));
+
+    const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!;
+    expect(forwarded.dataPoints.map((dp) => (dp.value as { sum: number }).sum)).toEqual([5]);
+  });
+
+  describe("against a real otel collector", () => {
+    let collector: StartedOtelCollectorContainer;
+
+    beforeAll(async () => {
+      collector = await new OtelCollectorContainer().start();
+    }, 180_000);
+
+    afterAll(async () => {
+      await collector?.stop();
+    });
+
+    it("exports a batch containing a NaN gauge value without being rejected", async () => {
+      const exporter = new BufferingMetricExporter(
+        new OTLPMetricExporter({ url: `${collector.getOtlpHttpUrl()}/v1/metrics` }),
+        30_000
+      );
+
+      exporter.export(resourceMetrics([gauge("process.cpu.utilization", [NaN, 0.25])]), () => {});
+
+      await expect(exporter.forceFlush()).resolves.toBeUndefined();
+    });
+  });
+});

packages/core/src/v3/taskContext/otelProcessors.ts

@@ -224,6 +224,37 @@ export class TaskContextMetricExporter implements PushMetricExporter {
   }
 }
 
+function isFiniteDataPointValue(value: unknown): boolean {
+  if (typeof value === "number") {
+    return Number.isFinite(value);
+  }
+
+  if (typeof value === "object" && value !== null) {
+    return (["sum", "min", "max"] as const).every((key) => {
+      const component = (value as Record<string, unknown>)[key];
+      return typeof component !== "number" || Number.isFinite(component);
+    });
+  }
+
+  return true;
+}
+
+function dropNonFiniteDataPoints(metrics: ResourceMetrics): ResourceMetrics {
+  return {
+    ...metrics,
+    scopeMetrics: metrics.scopeMetrics.map((scope) => ({
+      ...scope,
+      metrics: scope.metrics.map(
+        (metric) =>
+          ({
+            ...metric,
+            dataPoints: metric.dataPoints.filter((dp) => isFiniteDataPointValue(dp.value)),
+          }) as MetricData
+      ),
+    })),
+  };
+}
+
 export class BufferingMetricExporter implements PushMetricExporter {
   selectAggregationTemporality?: (instrumentType: InstrumentType) => AggregationTemporality;
   selectAggregation?: (instrumentType: InstrumentType) => AggregationOption;
@@ -245,7 +276,7 @@ export class BufferingMetricExporter implements PushMetricExporter {
   }
 
   export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void {
-    this._buffer.push(metrics);
+    this._buffer.push(dropNonFiniteDataPoints(metrics));
 
     const now = Date.now();
     if (now - this._lastFlushTime >= this._flushIntervalMs) {

pnpm-lock.yaml

@@ -1474,9 +1474,6 @@ importers:
       '@testcontainers/redis':
         specifier: ^11.14.0
         version: 11.14.0
-      '@trigger.dev/core':
-        specifier: workspace:*
-        version: link:../../packages/core
       std-env:
         specifier: ^3.9.0
         version: 3.9.0
@@ -1940,6 +1937,9 @@ importers:
       '@epic-web/test-server':
         specifier: ^0.1.0
         version: 0.1.0(bufferutil@4.0.9)
+      '@internal/testcontainers':
+        specifier: workspace:*
+        version: link:../../internal-packages/testcontainers
       '@trigger.dev/database':
         specifier: workspace:*
         version: link:../../internal-packages/database