github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

v3: superjson dynamic import and deploy fixes (#982)

Browse Source 
* v3: Dynamically import superjson and fix some bundling issues

* Added changeset

* Better error handling in the registry proxy and catch uncaught exceptions and unhandled promise rejections instead of crashing the server

* Await the prettyPrintPackage
This commit is contained in:
Eric Allam
2024-03-28 16:22:00 +00:00
committed by GitHub
parent a2365e406d
commit f93eae300e
14 changed files with 126 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/tall-bees-wave.md
+7
View File
@@ -0,0 +1,7 @@
---
"@trigger.dev/sdk": patch
"trigger.dev": patch
"@trigger.dev/core": patch
---
Dynamically import superjson and fix some bundling issues
apps/webapp/app/entry.server.tsx
+12 -4
View File
@@ -14,9 +14,9 @@ import {
OperatingSystemContextProvider,
OperatingSystemPlatform,
} from "./components/primitives/OperatingSystemProvider";
import { env } from "./env.server";
import { getSharedSqsEventConsumer } from "./services/events/sqsEventConsumer";
import { singleton } from "./utils/singleton";
import { logger } from "./services/logger.server";
const ABORT_DELAY = 30000;
@@ -178,7 +178,15 @@ function logError(error: unknown, request?: Request) {
const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);
export { wss } from "./v3/handleWebsockets.server";
export { socketIo } from "./v3/handleSocketIo.server";
export { registryProxy } from "./v3/registryProxy.server";
export { apiRateLimiter } from "./services/apiRateLimit.server";
export { socketIo } from "./v3/handleSocketIo.server";
export { wss } from "./v3/handleWebsockets.server";
export { registryProxy } from "./v3/registryProxy.server";
process.on("uncaughtException", (error, origin) => {
logger.error("Uncaught Exception", { error, origin });
});
process.on("unhandledRejection", (reason, promise) => {
logger.error("Unhandled Rejection", { reason });
});
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+2 -2
View File
@@ -43,14 +43,14 @@ export class SpanPresenter {
span.outputType === "application/store"
? `/resources/packets/${span.environmentId}/${span.output}`
: typeof span.output !== "undefined" && span.output !== null
? prettyPrintPacket(span.output, span.outputType ?? undefined)
? await prettyPrintPacket(span.output, span.outputType ?? undefined)
: undefined;
const payload =
span.payloadType === "application/store"
? `/resources/packets/${span.environmentId}/${span.payload}`
: typeof span.payload !== "undefined" && span.payload !== null
? prettyPrintPacket(span.payload, span.payloadType ?? undefined)
? await prettyPrintPacket(span.payload, span.payloadType ?? undefined)
: undefined;
return {
apps/webapp/app/v3/eventRepository.server.ts
+1 -1
View File
@@ -188,7 +188,7 @@ export class EventRepository {
const event = events[0];
const output = options?.attributes.output
? createPackageAttributesAsJson(
? await createPackageAttributesAsJson(
options?.attributes.output,
options?.attributes.outputType ?? "application/json"
)
apps/webapp/app/v3/registryProxy.server.ts
+26 -6
View File
@@ -259,6 +259,18 @@ export class RegistryProxy {
proxyRes.pipe(response, { end: true });
});
request.on("close", () => {
logger.debug("Client closed the connection");
proxyReq.destroy();
cleanupTempFile();
});
request.on("abort", () => {
logger.debug("Client aborted the connection");
proxyReq.destroy(); // Abort the proxied request
cleanupTempFile(); // Clean up the temporary file if necessary
});
if (tempFilePath) {
const readStream = createReadStream(tempFilePath);
@@ -427,14 +439,22 @@ function initializeProxy() {
});
}
async function streamRequestBodyToTempFile(request: IncomingMessage): Promise<string> {
const tempDir = await mkdtemp(`${tmpdir()}/`);
const tempFilePath = `${tempDir}/requestBody.tmp`;
const writeStream = createWriteStream(tempFilePath);
async function streamRequestBodyToTempFile(request: IncomingMessage): Promise<string | undefined> {
try {
const tempDir = await mkdtemp(`${tmpdir()}/`);
const tempFilePath = `${tempDir}/requestBody.tmp`;
const writeStream = createWriteStream(tempFilePath);
await pipeline(request, writeStream);
await pipeline(request, writeStream);
return tempFilePath;
return tempFilePath;
} catch (error) {
logger.error("Failed to stream request body to temp file", {
error: error instanceof Error ? error.message : error,
});
return;
}
}
type DockerImageParts = {
packages/cli-v3/src/commands/deploy.ts
+22 -3
View File
@@ -889,7 +889,14 @@ async function compileProject(
TRIGGER_API_URL: `"${config.triggerUrl}"`,
__PROJECT_CONFIG__: JSON.stringify(config),
},
plugins: [bundleDependenciesPlugin(config), workerSetupImportConfigPlugin(configPath)],
plugins: [
bundleDependenciesPlugin(
"workerFacade",
config.dependenciesToBundle,
config.tsconfigPath
),
workerSetupImportConfigPlugin(configPath),
],
});
if (result.errors.length > 0) {
@@ -927,15 +934,22 @@ async function compileProject(
write: false,
minify: false,
sourcemap: false,
packages: "external", // https://esbuild.github.io/api/#packages
logLevel: "error",
platform: "node",
packages: "external",
format: "cjs", // This is needed to support opentelemetry instrumentation that uses module patching
target: ["node18", "es2020"],
outdir: "out",
define: {
__PROJECT_CONFIG__: JSON.stringify(config),
},
plugins: [
bundleDependenciesPlugin(
"entryPoint.ts",
config.dependenciesToBundle,
config.tsconfigPath
),
],
});
if (entryPointResult.errors.length > 0) {
@@ -1003,6 +1017,11 @@ async function compileProject(
// Save the entryPoint outputFile to /tmp/dir/index.js
await writeFile(join(tempDir, "index.js"), entryPointOutputFile.text);
logger.debug("Getting the imports for the worker and entryPoint builds", {
workerImports: metaOutput.imports,
entryPointImports: entryPointMetaOutput.imports,
});
// Get all the required dependencies from the metaOutputs and save them to /tmp/dir/package.json
const allImports = [...metaOutput.imports, ...entryPointMetaOutput.imports];
@@ -1246,7 +1265,7 @@ async function gatherRequiredDependencies(
const dependencies: Record<string, string> = {};
for (const file of imports) {
if (file.kind !== "require-call" || !file.external) {
if ((file.kind !== "require-call" && file.kind !== "dynamic-import") || !file.external) {
continue;
}
packages/cli-v3/src/commands/dev.tsx
+10 -2
View File
@@ -362,7 +362,11 @@ function useDev({
__PROJECT_CONFIG__: JSON.stringify(config),
},
plugins: [
bundleDependenciesPlugin(config),
bundleDependenciesPlugin(
"workerFacade",
(config.dependenciesToBundle ?? []).concat([/^@trigger.dev/]),
config.tsconfigPath
),
workerSetupImportConfigPlugin(configPath),
{
name: "trigger.dev v3",
@@ -631,8 +635,12 @@ async function gatherRequiredDependencies(
) {
const dependencies: Record<string, string> = {};
logger.debug("Gathering required dependencies from imports", {
imports: outputMeta.imports,
});
for (const file of outputMeta.imports) {
if (file.kind !== "require-call" || !file.external) {
if ((file.kind !== "require-call" && file.kind !== "dynamic-import") || !file.external) {
continue;
}
packages/cli-v3/src/utilities/build.ts
+10 -32
View File
@@ -1,10 +1,9 @@
import { ResolvedConfig } from "@trigger.dev/core/v3";
import type * as esbuild from "esbuild";
import type { Plugin } from "esbuild";
import { readFileSync } from "node:fs";
import { extname, isAbsolute } from "node:path";
import tsConfigPaths from "tsconfig-paths";
import { logger } from "./logger";
import { readFileSync } from "node:fs";
export function workerSetupImportConfigPlugin(configPath?: string): Plugin {
return {
@@ -37,8 +36,12 @@ export function workerSetupImportConfigPlugin(configPath?: string): Plugin {
};
}
export function bundleDependenciesPlugin(config: ResolvedConfig): Plugin {
const matchPath = config.tsconfigPath ? createMatchPath(config.tsconfigPath) : undefined;
export function bundleDependenciesPlugin(
buildIdentifier: string,
dependenciesToBundle?: Array<string | RegExp>,
tsconfigPath?: string
): Plugin {
const matchPath = tsconfigPath ? createMatchPath(tsconfigPath) : undefined;
function resolvePath(id: string) {
if (!matchPath) {
@@ -53,33 +56,8 @@ export function bundleDependenciesPlugin(config: ResolvedConfig): Plugin {
build.onResolve({ filter: /.*/ }, (args) => {
const resolvedPath = resolvePath(args.path);
logger.ignore(`Checking if ${args.path} should be bundled or external`, {
...args,
resolvedPath,
});
if (!isBareModuleId(resolvedPath)) {
logger.ignore(`Bundling ${args.path} because its not a bareModuleId`, {
...args,
});
return undefined; // let esbuild bundle it
}
if (args.path.startsWith("@trigger.dev/")) {
logger.ignore(`Bundling ${args.path} because its a trigger.dev package`, {
...args,
});
return undefined; // let esbuild bundle it
}
if (args.path === "superjson" || args.path === "copy-anything" || args.path === "is-what") {
logger.debug(`Bundling ${args.path} because its superjson/copy-anything/is-what`, {
...args,
});
return undefined; // let esbuild bundle it
return undefined; // let esbuild handle it
}
// Skip assets that are treated as files (.css, .svg, .png, etc.).
@@ -97,13 +75,13 @@ export function bundleDependenciesPlugin(config: ResolvedConfig): Plugin {
return undefined;
}
for (let pattern of config.dependenciesToBundle ?? []) {
for (let pattern of dependenciesToBundle ?? []) {
if (typeof pattern === "string" ? args.path === pattern : pattern.test(args.path)) {
return undefined; // let esbuild bundle it
}
}
logger.ignore(`Externalizing ${args.path}`, {
logger.ignore(`[${buildIdentifier}] Externalizing ${args.path}`, {
...args,
});
packages/core/src/v3/utils/ioSerialization.ts
+24 -8
View File
@@ -1,5 +1,4 @@
import { Attributes, Span } from "@opentelemetry/api";
import { deserialize, parse, stringify } from "superjson";
import { apiClientManager } from "../apiClient";
import { OFFLOAD_IO_PACKET_LENGTH_LIMIT, imposeAttributeLimits } from "../limits";
import { SemanticInternalAttributes } from "../semanticInternalAttributes";
@@ -11,7 +10,7 @@ export type IOPacket = {
dataType: string;
};
export function parsePacket(value: IOPacket): any {
export async function parsePacket(value: IOPacket): Promise<any> {
if (!value.data) {
return undefined;
}
@@ -20,6 +19,8 @@ export function parsePacket(value: IOPacket): any {
case "application/json":
return JSON.parse(value.data);
case "application/super+json":
const { parse } = await loadSuperJSON();
return parse(value.data);
case "text/plain":
return value.data;
@@ -32,7 +33,7 @@ export function parsePacket(value: IOPacket): any {
}
}
export function stringifyIO(value: any): IOPacket {
export async function stringifyIO(value: any): Promise<IOPacket> {
if (value === undefined) {
return { dataType: "application/json" };
}
@@ -41,6 +42,8 @@ export function stringifyIO(value: any): IOPacket {
return { data: value, dataType: "text/plain" };
}
const { stringify } = await loadSuperJSON();
return { data: stringify(value), dataType: "application/super+json" };
}
@@ -186,11 +189,11 @@ async function importPacket(packet: IOPacket, span?: Span): Promise<IOPacket> {
return packet;
}
export function createPacketAttributes(
export async function createPacketAttributes(
packet: IOPacket,
dataKey: string,
dataTypeKey: string
): Attributes {
): Promise<Attributes> {
if (!packet.data) {
return {};
}
@@ -202,6 +205,8 @@ export function createPacketAttributes(
[dataTypeKey]: packet.dataType,
};
case "application/super+json":
const { parse } = await loadSuperJSON();
const parsed = parse(packet.data) as any;
const jsonified = JSON.parse(JSON.stringify(parsed, safeReplacer));
@@ -224,7 +229,10 @@ export function createPacketAttributes(
}
}
export function createPackageAttributesAsJson(data: any, dataType: string): Attributes {
export async function createPackageAttributesAsJson(
data: any,
dataType: string
): Promise<Attributes> {
if (
typeof data === "string" ||
typeof data === "number" ||
@@ -239,6 +247,8 @@ export function createPackageAttributesAsJson(data: any, dataType: string): Attr
case "application/json":
return imposeAttributeLimits(flattenAttributes(data, undefined));
case "application/super+json":
const { deserialize } = await loadSuperJSON();
const deserialized = deserialize(data) as any;
const jsonify = JSON.parse(JSON.stringify(deserialized, safeReplacer));
@@ -250,13 +260,15 @@ export function createPackageAttributesAsJson(data: any, dataType: string): Attr
}
}
export function prettyPrintPacket(rawData: any, dataType?: string): string {
export async function prettyPrintPacket(rawData: any, dataType?: string): Promise<string> {
if (rawData === undefined) {
return "";
}
if (dataType === "application/super+json") {
return prettyPrintPacket(deserialize(rawData), "application/json");
const { deserialize } = await loadSuperJSON();
return await prettyPrintPacket(deserialize(rawData), "application/json");
}
if (dataType === "application/json") {
@@ -310,3 +322,7 @@ function getPacketExtension(outputType: string): string {
return "txt";
}
}
async function loadSuperJSON(): Promise<typeof import("superjson")> {
return await import("superjson");
}
packages/core/src/v3/workers/taskExecutor.ts
+3 -3
View File
@@ -89,14 +89,14 @@ export class TaskExecutor {
try {
const payloadPacket = await conditionallyImportPacket(originalPacket, this._tracer);
parsedPayload = parsePacket(payloadPacket);
parsedPayload = await parsePacket(payloadPacket);
initOutput = await this.#callTaskInit(parsedPayload, ctx);
const output = await this.#callRun(parsedPayload, ctx, initOutput);
try {
const stringifiedOutput = stringifyIO(output);
const stringifiedOutput = await stringifyIO(output);
const finalOutput = await conditionallyExportPacket(
stringifiedOutput,
@@ -105,7 +105,7 @@ export class TaskExecutor {
);
span.setAttributes(
createPacketAttributes(
await createPacketAttributes(
finalOutput,
SemanticInternalAttributes.OUTPUT,
SemanticInternalAttributes.OUTPUT_TYPE
packages/trigger-sdk/src/v3/shared.ts
+1 -1
View File
@@ -551,7 +551,7 @@ async function handleTaskRunExecutionResult<TOutput>(
return {
ok: true,
id: execution.id,
output: parsePacket(importedPacket),
output: await parsePacket(importedPacket),
};
} else {
return {
pnpm-lock.yaml
Generated
+5 -87
View File
@@ -3050,9 +3050,12 @@ importers:
'@traceloop/instrumentation-openai':
specifier: ^0.3.9
version: 0.3.9(@opentelemetry/api@1.8.0)
'@trigger.dev/core':
specifier: workspace:^3.0.0-beta.0
version: link:../../packages/core
'@trigger.dev/sdk':
specifier: 2.3.18
version: 2.3.18
specifier: workspace:^3.0.0-beta.0
version: link:../../packages/trigger-sdk
msw:
specifier: ^2.2.1
version: 2.2.1(typescript@5.3.3)
@@ -3069,24 +3072,9 @@ importers:
'@types/node':
specifier: 20.4.2
version: 20.4.2
concurrently:
specifier: ^8.2.0
version: 8.2.0
dotenv:
specifier: ^16.3.1
version: 16.3.1
nodemon:
specifier: ^3.0.1
version: 3.0.1
trigger.dev:
specifier: workspace:*
version: link:../../packages/cli-v3
ts-node:
specifier: ^10.9.1
version: 10.9.1(@types/node@20.4.2)(typescript@5.3.3)
tsconfig-paths:
specifier: ^3.14.1
version: 3.14.1
typescript:
specifier: ^5.3.0
version: 5.3.3
@@ -13115,11 +13103,6 @@ packages:
react-dom: 18.2.0(react@18.2.0)
dev: false
/@trigger.dev/core-backend@2.3.18:
resolution: {integrity: sha512-LVeeerraGeqKNd2gtajQY+mnGWqkYW7Q2r5oWpL5xIZ8aQg3HRhSIfZs1dryexwKlfqnRjGWueGTy2+j1tbzcg==}
engines: {node: '>=18.0.0'}
dev: false
/@trigger.dev/core@2.0.7:
resolution: {integrity: sha512-z86G0sbqu0ePP0eg3jA/v65TOq9ZLCHUpgis/cqaLopsybnMScWt9S9MV1Dq9FVqHHqKvyuLBugZTJwsj+FXGg==}
engines: {node: '>=16.8.0'}
@@ -13129,15 +13112,6 @@ packages:
zod-error: 1.5.0
dev: false
/@trigger.dev/core@2.3.18:
resolution: {integrity: sha512-j2EdCeyMkZ+zlVnnHl5zmBb+YURSw4x75NqQU1G5X08pQAza7G0qEn8DDGIMR5ieUMiHP0WS9oYy/voYdNfibQ==}
engines: {node: '>=18.0.0'}
dependencies:
ulidx: 2.2.1
zod: 3.22.3
zod-error: 1.5.0
dev: false
/@trigger.dev/nextjs@1.0.0(@trigger.dev/sdk@2.0.7)(next@13.4.12):
resolution: {integrity: sha512-Dt32BaAKYNJqYbPcpdzUOfkgpDRIftyd4oj2lBFilIHZUCJpfG1MdYbr4YWFhNE5Z04Pj0I+uN8kaxTkTZp+Ig==}
engines: {node: '>=16.8.0'}
@@ -13179,31 +13153,6 @@ packages:
- utf-8-validate
dev: false
/@trigger.dev/sdk@2.3.18:
resolution: {integrity: sha512-Bjxgl4BbWOAL8rhxeBkl7SzvLLRBMJjiftq/7W7u96MDyPRFUoZZvVMSZzTJufnLBf/xS2JTi8LWU8gzhDJDvw==}
engines: {node: '>=18.0.0'}
dependencies:
'@trigger.dev/core': 2.3.18
'@trigger.dev/core-backend': 2.3.18
chalk: 5.3.0
cronstrue: 2.21.0
debug: 4.3.4(supports-color@8.1.1)
evt: 2.5.7
get-caller-file: 2.0.5
git-remote-origin-url: 4.0.0
git-repo-info: 2.1.1
slug: 6.1.0
terminal-link: 3.0.0
ulid: 2.3.0
uuid: 9.0.0
ws: 8.16.0
zod: 3.22.3
transitivePeerDependencies:
- bufferutil
- supports-color
- utf-8-validate
dev: false
/@tsconfig/node10@1.0.9:
resolution: {integrity: sha512-jNsYVVxU8v5g43Erja32laIDHXeoNvFEpX33OK4d6hljo3jDhCBDhx5dhCCTMWUojscpAagGiRkBKxpdl9fxqA==}
@@ -31044,37 +30993,6 @@ packages:
yn: 3.1.1
dev: true
/ts-node@10.9.1(@types/node@20.4.2)(typescript@5.3.3):
resolution: {integrity: sha512-NtVysVPkxxrwFGUUxGYhfux8k78pQB3JqYBXlLRZgdGUqTO5wU/UyHop5p70iEbGhB7q5KmiZiU0Y3KlJrScEw==}
hasBin: true
peerDependencies:
'@swc/core': '>=1.2.50'
'@swc/wasm': '>=1.2.50'
'@types/node': '*'
typescript: '>=2.7'
peerDependenciesMeta:
'@swc/core':
optional: true
'@swc/wasm':
optional: true
dependencies:
'@cspotcode/source-map-support': 0.8.1
'@tsconfig/node10': 1.0.9
'@tsconfig/node12': 1.0.11
'@tsconfig/node14': 1.0.3
'@tsconfig/node16': 1.0.3
'@types/node': 20.4.2
acorn: 8.10.0
acorn-walk: 8.2.0
arg: 4.1.3
create-require: 1.1.1
diff: 4.0.2
make-error: 1.3.6
typescript: 5.3.3
v8-compile-cache-lib: 3.0.1
yn: 3.1.1
dev: true
/ts-node@10.9.1(@types/node@20.5.0)(typescript@5.1.6):
resolution: {integrity: sha512-NtVysVPkxxrwFGUUxGYhfux8k78pQB3JqYBXlLRZgdGUqTO5wU/UyHop5p70iEbGhB7q5KmiZiU0Y3KlJrScEw==}
hasBin: true
references/v3-catalog/package.json
+2 -1
View File
@@ -9,7 +9,8 @@
"@opentelemetry/api": "^1.8.0",
"@sindresorhus/slugify": "^2.2.1",
"@traceloop/instrumentation-openai": "^0.3.9",
"@trigger.dev/sdk": "2.3.18",
"@trigger.dev/sdk": "workspace:^3.0.0-beta.0",
"@trigger.dev/core": "workspace:^3.0.0-beta.0",
"msw": "^2.2.1",
"openai": "^4.28.0",
"stripe": "^12.14.0"
references/v3-catalog/src/trigger/superjson.ts
+1 -1
View File
@@ -30,7 +30,7 @@ export const superChildTask = task({
date: new Date(),
regex: /foo/,
bigint: BigInt(123),
set: new Set([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]),
set: new Set([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]),
map: new Map([
["foo", "bar"],
["baz", "qux"],