Merge branch 'main' into fix/sdk-stream-root-fallback

Author: bharathkumar39293

.changeset/ai-sdk-v6-support.md

@@ -201,6 +201,7 @@ function ReplayForm({
       tags,
       version,
       machine,
+      prioritySeconds,
     },
   ] = useForm({
     id: "replay-task",
@@ -499,6 +500,12 @@ function ReplayForm({
                 <Hint>Delays run by a specific duration.</Hint>
                 <FormError id={delaySeconds.errorId}>{delaySeconds.error}</FormError>
               </InputGroup>
+              <InputGroup>
+                <Label variant="small">Priority</Label>
+                <DurationPicker name={prioritySeconds.name} id={prioritySeconds.id} />
+                <Hint>Sets the priority of the run. Higher values mean higher priority.</Hint>
+                <FormError id={prioritySeconds.errorId}>{prioritySeconds.error}</FormError>
+              </InputGroup>
               <InputGroup>
                 <Label variant="small">TTL</Label>
                 <DurationPicker

.changeset/bright-keys-shine.md

@@ -1,6 +1,6 @@
 import { nanoid, customAlphabet } from "nanoid";
 import slug from "slug";
-import { prisma } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import type { Project } from "@trigger.dev/database";
 import { Organization, createEnvironment } from "./organization.server";
 import { env } from "~/env.server";
@@ -135,7 +135,7 @@ export async function createProject(
 
 export async function findProjectBySlug(orgSlug: string, projectSlug: string, userId: string) {
   // Find the project scoped to the organization, making sure the user belongs to that org
-  return await prisma.project.findFirst({
+  return await $replica.project.findFirst({
     where: {
       slug: projectSlug,
       organization: {
@@ -148,7 +148,7 @@ export async function findProjectBySlug(orgSlug: string, projectSlug: string, us
 
 export async function findProjectByRef(externalRef: string, userId: string) {
   // Find the project scoped to the organization, making sure the user belongs to that org
-  return await prisma.project.findFirst({
+  return await $replica.project.findFirst({
     where: {
       externalRef,
       organization: {

.changeset/gentle-streams-flow.md

@@ -1,6 +1,6 @@
 import type { AuthenticatedEnvironment } from "@internal/run-engine";
 import type { Prisma, PrismaClientOrTransaction, RuntimeEnvironment } from "@trigger.dev/database";
-import { prisma } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { getUsername } from "~/utils/username";
 import { sanitizeBranchName } from "~/v3/gitBranch";
@@ -11,7 +11,7 @@ export async function findEnvironmentByApiKey(
   apiKey: string,
   branchName: string | undefined
 ): Promise<AuthenticatedEnvironment | null> {
-  const environment = await prisma.runtimeEnvironment.findFirst({
+  const environment = await $replica.runtimeEnvironment.findFirst({
     where: {
       apiKey,
     },
@@ -67,7 +67,7 @@ export async function findEnvironmentByPublicApiKey(
   apiKey: string,
   branchName: string | undefined
 ): Promise<AuthenticatedEnvironment | null> {
-  const environment = await prisma.runtimeEnvironment.findFirst({
+  const environment = await $replica.runtimeEnvironment.findFirst({
     where: {
       pkApiKey: apiKey,
     },
@@ -89,7 +89,7 @@ export async function findEnvironmentByPublicApiKey(
 export async function findEnvironmentById(
   id: string
 ): Promise<(AuthenticatedEnvironment & { parentEnvironment: { apiKey: string } | null }) | null> {
-  const environment = await prisma.runtimeEnvironment.findFirst({
+  const environment = await $replica.runtimeEnvironment.findFirst({
     where: {
       id,
     },
@@ -118,7 +118,7 @@ export async function findEnvironmentBySlug(
   envSlug: string,
   userId: string
 ): Promise<AuthenticatedEnvironment | null> {
-  return prisma.runtimeEnvironment.findFirst({
+  return $replica.runtimeEnvironment.findFirst({
     where: {
       projectId: projectId,
       slug: envSlug,
@@ -148,7 +148,7 @@ export async function findEnvironmentFromRun(
   runId: string,
   tx?: PrismaClientOrTransaction
 ): Promise<AuthenticatedEnvironment | null> {
-  const taskRun = await (tx ?? prisma).taskRun.findFirst({
+  const taskRun = await (tx ?? $replica).taskRun.findFirst({
     where: {
       id: runId,
     },
@@ -223,7 +223,7 @@ export async function disconnectSession(environmentId: string) {
 }
 
 export async function findLatestSession(environmentId: string) {
-  const session = await prisma.runtimeEnvironmentSession.findFirst({
+  const session = await $replica.runtimeEnvironmentSession.findFirst({
     where: {
       environmentId,
     },
@@ -280,7 +280,7 @@ export async function findDisplayableEnvironment(
   environmentId: string,
   userId: string | undefined
 ) {
-  const environment = await prisma.runtimeEnvironment.findFirst({
+  const environment = await $replica.runtimeEnvironment.findFirst({
     where: {
       id: environmentId,
     },

.changeset/thirty-trainers-reflect.md

@@ -83,10 +83,12 @@ export class LimitsPresenter extends BasePresenter {
   public async call({
     organizationId,
     projectId,
+    environmentId,
     environmentApiKey,
   }: {
     organizationId: string;
     projectId: string;
+    environmentId: string;
     environmentApiKey: string;
   }): Promise<LimitsResult> {
     // Get organization with all limit-related fields
@@ -159,9 +161,9 @@ export class LimitsPresenter extends BasePresenter {
       environmentApiKey,
       apiRateLimitConfig
     );
-    const batchRateLimitTokens = await getRateLimitRemainingTokens(
-      "batch",
-      environmentApiKey,
+    // Batch rate limiter uses environment ID directly (not hashed) with a different key prefix
+    const batchRateLimitTokens = await getBatchRateLimitRemainingTokens(
+      environmentId,
       batchRateLimitConfig
     );
 
@@ -419,3 +421,36 @@ async function getRateLimitRemainingTokens(
     return null;
   }
 }
+
+/**
+ * Query the current remaining tokens for the batch rate limiter.
+ * The batch rate limiter uses environment ID directly (not hashed) and has a different key prefix.
+ */
+async function getBatchRateLimitRemainingTokens(
+  environmentId: string,
+  config: RateLimiterConfig
+): Promise<number | null> {
+  try {
+    // Create a Ratelimit instance with the same configuration as the batch rate limiter
+    const limiter = createLimiterFromConfig(config);
+    const ratelimit = new Ratelimit({
+      redis: rateLimitRedisClient,
+      limiter,
+      ephemeralCache: new Map(),
+      analytics: false,
+      // The batch rate limiter uses "ratelimit:batch" as keyPrefix in RateLimiter,
+      // which adds another "ratelimit:" prefix, resulting in "ratelimit:ratelimit:batch"
+      prefix: `ratelimit:ratelimit:batch`,
+    });
+
+    // Batch rate limiter uses environment ID directly (not hashed)
+    const remaining = await ratelimit.getRemaining(environmentId);
+    return remaining;
+  } catch (error) {
+    logger.warn("Failed to get batch rate limit remaining tokens", {
+      environmentId,
+      error: error instanceof Error ? error.message : String(error),
+    });
+    return null;
+  }
+}

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

@@ -81,6 +81,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     presenter.call({
       organizationId: project.organizationId,
       projectId: project.id,
+      environmentId: environment.id,
       environmentApiKey: environment.apiKey,
     })
   );

apps/webapp/app/models/project.server.ts

@@ -409,6 +409,7 @@ function StandardTaskForm({
       tags,
       version,
       machine,
+      prioritySeconds,
     },
   ] = useForm({
     id: "test-task",
@@ -730,6 +731,12 @@ function StandardTaskForm({
                 <Hint>Delays run by a specific duration.</Hint>
                 <FormError id={delaySeconds.errorId}>{delaySeconds.error}</FormError>
               </InputGroup>
+              <InputGroup>
+                <Label variant="small">Priority</Label>
+                <DurationPicker name={prioritySeconds.name} id={prioritySeconds.id} />
+                <Hint>Sets the priority of the run. Higher values mean higher priority.</Hint>
+                <FormError id={prioritySeconds.errorId}>{prioritySeconds.error}</FormError>
+              </InputGroup>
               <InputGroup>
                 <Label variant="small">TTL</Label>
                 <DurationPicker
@@ -872,6 +879,7 @@ function ScheduledTaskForm({
       tags,
       version,
       machine,
+      prioritySeconds,
     },
   ] = useForm({
     id: "test-task-scheduled",
@@ -1237,6 +1245,14 @@ function ScheduledTaskForm({
             <Hint>Limits concurrency by creating a separate queue for each value of the key.</Hint>
             <FormError id={concurrencyKey.errorId}>{concurrencyKey.error}</FormError>
           </InputGroup>
+          <InputGroup>
+            <Label htmlFor={prioritySeconds.id} variant="small">
+              Priority
+            </Label>
+            <DurationPicker name={prioritySeconds.name} id={prioritySeconds.id} />
+            <Hint>Sets the priority of the run. Higher values mean higher priority.</Hint>
+            <FormError id={prioritySeconds.errorId}>{prioritySeconds.error}</FormError>
+          </InputGroup>
           <InputGroup>
             <Label htmlFor={ttlSeconds.id} variant="small">
               TTL