a999d9ea3f
New batch trigger system with larger payloads, streaming ingestion, larger batch sizes, and a fair processing system. This PR introduces a new `FairQueue` abstraction inspired by our own `RunQueue` that enables multi-tenant fair queueing with concurrency limits. The new `BatchQueue` is built on top of the `FairQueue`, and handles processing Batch triggers in a fair manner with per-environment concurrency limits defined per-org. Additionally, there is a global concurrency limit to prevent the BatchQueue system from creating too many runs too quickly, which can cause downstream issues. For this new BatchQueue system we have a completely new batch trigger creation and ingestion system. Previously this was a single endpoint with a single JSON body that defined details about the batch as well as all the items in the batch. We're introducing a two-phase batch trigger ingestion system. In the first phase, the BatchTaskRun record is created (and possibly rate limited). The second phase is another endpoint that accepts an NDJSON body with each line being a single item/run with payload and options. At ingestion time all items are added to a queue, in order, and then processed by the BatchQueue system. ## New batch trigger rate limits This PR implements a new batch trigger specific rate limit, configured on the `Organization.batchRateLimitConfig` column, and defaults using these environment variables: - `BATCH_RATE_LIMIT_REFILL_RATE` defaults to 10 - `BATCH_RATE_LIMIT_REFILL_INTERVAL` the duration interval, defaults to `"10s"` - `BATCH_RATE_LIMIT_MAX` defaults to 1200 This rate limiter is scoped to the environment ID and controls how many runs can be submitted via batch triggers per interval. The SDK handles the retrying side. ## Batch queue concurrency limits The new column `Organization.batchQueueConcurrencyConfig` now defines an org specific `processingConcurrency` value, with a backup of the env var `BATCH_CONCURRENCY_LIMIT_DEFAULT` which defaults to 10. This controls how many batch queue items are processed concurrently per environment. There is also a global rate limit for the batch queue set via the `BATCH_QUEUE_GLOBAL_RATE_LIMIT` which defaults to being disabled. If set, the entire batch queue system won't process more than `BATCH_QUEUE_GLOBAL_RATE_LIMIT` items per second. This allows controlling the maximum number of runs created per second via batch triggers. ## Batch trigger settings - `STREAMING_BATCH_MAX_ITEMS` controls the maximum number of items in a single batch - `STREAMING_BATCH_ITEM_MAXIMUM_SIZE` controls the maximum size of each item in a batch - `BATCH_CONCURRENCY_DEFAULT_CONCURRENCY` controls the default environment concurrency - `BATCH_QUEUE_DRR_QUANTUM` how many credits each environment gets each round for the DRR scheduler - `BATCH_QUEUE_MAX_DEFICIT` the maximum deficit for the DRR scheduler - `BATCH_QUEUE_CONSUMER_COUNT` how many queue consumers to run - `BATCH_QUEUE_CONSUMER_INTERVAL_MS` how frequently they poll for items in the queue ### Configuration Recommendations by Use Case **High-throughput priority (fairness acceptable at 0.98+):** ```env BATCH_QUEUE_DRR_QUANTUM=25 BATCH_QUEUE_MAX_DEFICIT=100 BATCH_QUEUE_CONSUMER_COUNT=10 BATCH_QUEUE_CONSUMER_INTERVAL_MS=50 BATCH_CONCURRENCY_DEFAULT_CONCURRENCY=25 ``` **Strict fairness priority (throughput can be lower):** ```env BATCH_QUEUE_DRR_QUANTUM=5 BATCH_QUEUE_MAX_DEFICIT=25 BATCH_QUEUE_CONSUMER_COUNT=3 BATCH_QUEUE_CONSUMER_INTERVAL_MS=100 BATCH_CONCURRENCY_DEFAULT_CONCURRENCY=5 ```
258 lines
7.5 KiB
TypeScript
258 lines
7.5 KiB
TypeScript
import { createReadableStreamFromReadable, type EntryContext } from "@remix-run/node"; // or cloudflare/deno
|
|
import { RemixServer } from "@remix-run/react";
|
|
import { wrapHandleErrorWithSentry } from "@sentry/remix";
|
|
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 { 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";
|
|
|
|
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;
|
|
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);
|
|
},
|
|
onShellError(error: unknown) {
|
|
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);
|
|
}
|
|
},
|
|
}
|
|
);
|
|
|
|
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;
|
|
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);
|
|
},
|
|
onShellError(error: unknown) {
|
|
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);
|
|
}
|
|
},
|
|
}
|
|
);
|
|
|
|
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);
|
|
});
|
|
|
|
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);
|
|
|
|
export { apiRateLimiter } from "./services/apiRateLimit.server";
|
|
export { engineRateLimiter } from "./services/engineRateLimit.server";
|
|
export { runWithHttpContext } from "./services/httpAsyncStorage.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);
|
|
}
|