Files
triggerdotdev--trigger.dev/apps/webapp/app/entry.server.tsx
Eric Allam f9d57d3bd5 feat(webapp): add a new backend for the realtime runs feed (#3864)
## Summary

Adds a second backend for the realtime runs feed (`useRealtimeRun`,
`subscribeToRunsWithTag`, `subscribeToBatch`), built to stay healthy
when a single busy environment has many subscribers watching many runs
at once. It is gated behind a feature flag with the existing backend as
the default, so nothing changes for users until it is enabled per
environment.

## Design

A run change is published once, as a small self-describing record, to a
single per-environment channel. Every feed is then a predicate over that
one stream rather than owning a channel:

- A per-instance router indexes the currently-held feeds by run, tag,
and batch. When a run changes it hydrates the affected rows once and
serializes them once, then fans the result to every matching feed. One
hot shared tag watched by many subscribers costs a single database query
and serialize, not one per subscriber.
- Feeds that don't match a change are never woken, wake delivery per
environment is coalesced on a leading edge (250ms default) so a burst of
changes costs one wake, and cold reads coalesce onto a single
short-TTL-cached resolve.
- An admission gate bounds how many cold ClickHouse resolves run
concurrently, so a mass reconnect across many distinct filters queues
instead of stampeding the database.
- Changes that land while a client is between long-polls are delivered
on its next poll instead of waiting for the periodic backstop: each
environment buffers its recent change records, subscriptions linger
briefly after the last feed closes, and a newly-armed poll replays
exactly the connection's gap.
- The per-connection replay cursors behind that are shared across
instances via Redis (a single timestamp each), so a poll landing on a
different instance behind the load balancer still reads the connection's
true gap instead of falling back to a cold resolve. Cursor reads have a
bounded deadline and degrade to the cold-read path on any Redis trouble.
- Tag subscriptions with multiple tags match runs carrying all of the
tags, mirroring the existing backend's filter semantics, and live
long-polls hold for about 20 seconds to match its cadence.
- The per-environment channel supports Redis Cluster sharded pub/sub, so
the wake path scales horizontally across shards by environment.
- The backend reports its health through OpenTelemetry metrics (delivery
lag, poll resolution paths, backstop outcomes, replay and cursor-store
activity), with a provisioned Grafana dashboard for local development.

Everything is behind the feature flag and tunable via env vars; the
existing backend remains the default.
2026-06-11 07:56:10 +01:00

311 lines
10 KiB
TypeScript

import { createReadableStreamFromReadable, type EntryContext } from "@remix-run/node"; // or cloudflare/deno
import { RemixServer } from "@remix-run/react";
import * as Sentry from "@sentry/remix";
import { wrapHandleErrorWithSentry } from "@sentry/remix";
import { addTenantContextToEvent } from "~/utils/sentryTenantContext.server";
import { parseAcceptLanguage } from "intl-parse-accept-language";
import isbot from "isbot";
import { renderToPipeableStream } from "react-dom/server";
import { PassThrough } from "stream";
import * as Worker from "~/services/worker.server";
import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
import { bootstrap } from "./bootstrap";
import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
import {
OperatingSystemContextProvider,
OperatingSystemPlatform,
} from "./components/primitives/OperatingSystemProvider";
import { Prisma } from "./db.server";
import { env } from "./env.server";
import { eventLoopMonitor } from "./eventLoopMonitor.server";
import { logger } from "./services/logger.server";
import { resourceMonitor } from "./services/resourceMonitor.server";
import { singleton } from "./utils/singleton";
import { remoteBuildsEnabled } from "./v3/remoteImageBuilder.server";
import {
registerRunEngineEventBusHandlers,
setupBatchQueueCallbacks,
} from "./v3/runEngineHandlers.server";
import { registerRunChangeNotifierHandlers } from "./services/realtime/runChangeNotifierHandlers.server";
// Touch the sessions replication singleton at entry so it boots deterministically
// on webapp startup. The singleton's initializer wires start (gated on
// `clickhouseFactory.isReady()`) and SIGTERM/SIGINT shutdown — mirrors
// runsReplicationInstance.
//
// IMPORTANT: do NOT replace this with `void sessionsReplicationInstance;`.
// `apps/webapp/package.json` declares `"sideEffects": false`, so esbuild
// treats `void <identifier>;` as a pure expression statement and tree-shakes
// the entire import — the singleton's initializer never fires and the
// sessions→ClickHouse logical replication slot stops being consumed. Assigning
// to globalThis is an unambiguous side effect the bundler must preserve. See
// TRI-9864 for the incident write-up.
import { sessionsReplicationInstance } from "./services/sessionsReplicationInstance.server";
(globalThis as Record<string, unknown>).__sessionsReplicationInstance =
sessionsReplicationInstance;
const ABORT_DELAY = 30000;
export default function handleRequest(
request: Request,
responseStatusCode: number,
responseHeaders: Headers,
remixContext: EntryContext
) {
const url = new URL(request.url);
if (url.pathname.startsWith("/login")) {
responseHeaders.set("X-Frame-Options", "SAMEORIGIN");
responseHeaders.set("Content-Security-Policy", "frame-ancestors 'self'");
}
const acceptLanguage = request.headers.get("accept-language");
const locales = parseAcceptLanguage(acceptLanguage, {
validate: Intl.DateTimeFormat.supportedLocalesOf,
});
//get whether it's a mac or pc from the headers
const platform: OperatingSystemPlatform = request.headers.get("user-agent")?.includes("Mac")
? "mac"
: "windows";
// If the request is from a bot, we want to wait for the full
// response to render before sending it to the client. This
// ensures that bots can see the full page content.
if (isbot(request.headers.get("user-agent"))) {
return handleBotRequest(
request,
responseStatusCode,
responseHeaders,
remixContext,
locales,
platform
);
}
return handleBrowserRequest(
request,
responseStatusCode,
responseHeaders,
remixContext,
locales,
platform
);
}
function handleBotRequest(
request: Request,
responseStatusCode: number,
responseHeaders: Headers,
remixContext: EntryContext,
locales: string[],
platform: OperatingSystemPlatform
) {
return new Promise((resolve, reject) => {
let shellRendered = false;
// Timer handle is cleared in every terminal callback so the abort closure
// (which captures the full React render tree + remixContext) doesn't pin
// memory for 30s per successful request. See react-router PR #14200.
let abortTimer: NodeJS.Timeout | undefined;
const { pipe, abort } = renderToPipeableStream(
<OperatingSystemContextProvider platform={platform}>
<LocaleContextProvider locales={locales}>
<RemixServer context={remixContext} url={request.url} abortDelay={ABORT_DELAY} />,
</LocaleContextProvider>
</OperatingSystemContextProvider>,
{
onAllReady() {
shellRendered = true;
const body = new PassThrough();
const stream = createReadableStreamFromReadable(body);
responseHeaders.set("Content-Type", "text/html");
resolve(
new Response(stream, {
headers: responseHeaders,
status: responseStatusCode,
})
);
pipe(body);
clearTimeout(abortTimer);
},
onShellError(error: unknown) {
clearTimeout(abortTimer);
reject(error);
},
onError(error: unknown) {
responseStatusCode = 500;
// Log streaming rendering errors from inside the shell. Don't log
// errors encountered during initial shell rendering since they'll
// reject and get logged in handleDocumentRequest.
if (shellRendered) {
console.error(error);
}
},
}
);
abortTimer = setTimeout(abort, ABORT_DELAY);
});
}
function handleBrowserRequest(
request: Request,
responseStatusCode: number,
responseHeaders: Headers,
remixContext: EntryContext,
locales: string[],
platform: OperatingSystemPlatform
) {
return new Promise((resolve, reject) => {
let shellRendered = false;
// Timer handle is cleared in every terminal callback so the abort closure
// (which captures the full React render tree + remixContext) doesn't pin
// memory for 30s per successful request. See react-router PR #14200.
let abortTimer: NodeJS.Timeout | undefined;
const { pipe, abort } = renderToPipeableStream(
<OperatingSystemContextProvider platform={platform}>
<LocaleContextProvider locales={locales}>
<RemixServer context={remixContext} url={request.url} abortDelay={ABORT_DELAY} />
</LocaleContextProvider>
</OperatingSystemContextProvider>,
{
onShellReady() {
shellRendered = true;
const body = new PassThrough();
const stream = createReadableStreamFromReadable(body);
responseHeaders.set("Content-Type", "text/html");
resolve(
new Response(stream, {
headers: responseHeaders,
status: responseStatusCode,
})
);
pipe(body);
clearTimeout(abortTimer);
},
onShellError(error: unknown) {
clearTimeout(abortTimer);
reject(error);
},
onError(error: unknown) {
responseStatusCode = 500;
// Log streaming rendering errors from inside the shell. Don't log
// errors encountered during initial shell rendering since they'll
// reject and get logged in handleDocumentRequest.
if (shellRendered) {
console.error(error);
}
},
}
);
abortTimer = setTimeout(abort, ABORT_DELAY);
});
}
export const handleError = wrapHandleErrorWithSentry((error, { request }) => {
if (request instanceof Request) {
logger.debug("Error in handleError", {
error,
request: {
url: request.url,
method: request.method,
},
});
} else {
logger.debug("Error in handleError", {
error,
});
}
});
Worker.init().catch((error) => {
logError(error);
});
initMollifierDrainerWorker();
initMollifierStaleSweepWorker();
bootstrap().catch((error) => {
logError(error);
});
function logError(error: unknown, request?: Request) {
console.error(error);
if (error instanceof Error && error.message.startsWith("There are locked jobs present")) {
console.log("⚠️ graphile-worker migration issue detected!");
}
}
process.on("uncaughtException", (error, origin) => {
if (
error instanceof Prisma.PrismaClientKnownRequestError ||
error instanceof Prisma.PrismaClientUnknownRequestError
) {
// Don't exit the process if the error is a Prisma error
logger.error("uncaughtException prisma error", {
error,
prismaMessage: error.message,
code: "code" in error ? error.code : undefined,
meta: "meta" in error ? error.meta : undefined,
stack: error.stack,
origin,
});
} else {
logger.error("uncaughtException", {
error: { name: error.name, message: error.message, stack: error.stack },
origin,
});
}
process.exit(1);
});
singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks);
// Attach the realtime run-changed publish delegations to the engine event bus.
// No-ops (registers nothing) unless REALTIME_BACKEND_NATIVE_ENABLED=1.
singleton("RunChangeNotifierHandlers", registerRunChangeNotifierHandlers);
// Wrapped in singleton() so Remix's dev-mode CJS reloads don't append
// duplicate copies of the processor — Sentry's processor list lives in
// node_modules and persists across module reloads. Idempotent at runtime
// (the processor is a pure read+stamp), but the pattern matches the rest
// of this file.
singleton("SentryTenantContextProcessor", () => {
if (env.SENTRY_DSN) {
Sentry.addEventProcessor(addTenantContextToEvent);
}
// Return a truthy value — `singleton()` uses `??=` so a `void`
// callback would re-execute (and re-register) on every dev reload.
return true;
});
export { apiRateLimiter } from "./services/apiRateLimit.server";
export { engineRateLimiter } from "./services/engineRateLimit.server";
export { runWithHttpContext } from "./services/httpAsyncStorage.server";
export { tenantContextMiddleware } from "./services/tenantContextResolver.server";
export { socketIo } from "./v3/handleSocketIo.server";
export { wss } from "./v3/handleWebsockets.server";
if (env.EVENT_LOOP_MONITOR_ENABLED === "1") {
eventLoopMonitor.enable();
}
if (remoteBuildsEnabled()) {
console.log("🏗️ Remote builds enabled");
} else {
console.log("🏗️ Local builds enabled");
}
if (env.RESOURCE_MONITOR_ENABLED === "1") {
resourceMonitor.startMonitoring(1000);
}