address comments

Author: icecrasher321

apps/sim/app/api/schedules/execute/route.ts

@@ -41,7 +41,9 @@ const JOB_CHUNK_SIZE = 100
 const MAX_TICK_DURATION_MS = 3 * 60 * 1000
 const STALE_SCHEDULE_CLAIM_MS = getMaxExecutionTimeout()
 const STALE_SCHEDULE_RECOVERY_BATCH_SIZE = 100
-let databaseScheduleStartQueue: Promise<void> = Promise.resolve()
+const DATABASE_SCHEDULE_START_TURN_WAIT_MS = 1_000
+type DatabaseScheduleStartResult = 'started' | 'capacity_full' | 'not_pending'
+let databaseScheduleStartTurn: Promise<void> | null = null
 
 const dueFilter = (queuedAt: Date) =>
   and(
@@ -68,18 +70,30 @@ const workflowScheduleFilter = (queuedAt: Date) =>
 const jobScheduleFilter = (queuedAt: Date) =>
   and(dueFilter(queuedAt), sql`${workflowSchedule.sourceType} = 'job'`)
 
-async function runWithDatabaseScheduleStartTurn<T>(operation: () => Promise<T>): Promise<T> {
-  const previousTurn = databaseScheduleStartQueue
+async function runWithDatabaseScheduleStartTurn(
+  operation: () => Promise<DatabaseScheduleStartResult>
+): Promise<DatabaseScheduleStartResult> {
+  const activeTurn = databaseScheduleStartTurn
+  if (activeTurn) {
+    const turnOpened = await Promise.race([
+      activeTurn.then(() => true),
+      sleep(DATABASE_SCHEDULE_START_TURN_WAIT_MS).then(() => false),
+    ])
+    if (!turnOpened || databaseScheduleStartTurn) return 'capacity_full'
+  }
+
   let releaseTurn = () => {}
-  databaseScheduleStartQueue = new Promise<void>((resolve) => {
+  const currentTurn = new Promise<void>((resolve) => {
     releaseTurn = resolve
   })
-
-  await previousTurn
+  databaseScheduleStartTurn = currentTurn
 
   try {
     return await operation()
   } finally {
+    if (databaseScheduleStartTurn === currentTurn) {
+      databaseScheduleStartTurn = null
+    }
     releaseTurn()
   }
 }
@@ -286,14 +300,10 @@ function staleScheduleExecutionJobsFilter(staleStartedBefore: Date) {
   )
 }
 
-function getExhaustedRecoveryNextRunAt(payload: ScheduleExecutionPayload, now: Date): Date {
-  return (
-    getNextRunFromCronExpression(payload.cronExpression, payload.timezone) ??
-    new Date(now.getTime() + 24 * 60 * 60 * 1000)
-  )
-}
-
-function getScheduleFailureNextRunAt(schedule: DatabaseScheduleExecutionTarget, now: Date): Date {
+function getScheduleNextRunAt(
+  schedule: { cronExpression?: string | null; timezone?: string },
+  now: Date
+): Date {
   return (
     getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone) ??
     new Date(now.getTime() + 24 * 60 * 60 * 1000)
@@ -313,7 +323,7 @@ async function markClaimedScheduleFailed(
       updatedAt: now,
       lastQueuedAt: null,
       lastFailedAt: now,
-      nextRunAt: getScheduleFailureNextRunAt(schedule, now),
+      nextRunAt: getScheduleNextRunAt(schedule, now),
       failedCount: sql`COALESCE(${workflowSchedule.failedCount}, 0) + 1`,
       status: sql`CASE WHEN COALESCE(${workflowSchedule.failedCount}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN 'disabled' ELSE 'active' END`,
       infraRetryCount: 0,
@@ -447,7 +457,7 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
           updatedAt: now,
           lastQueuedAt: null,
           lastFailedAt: now,
-          nextRunAt: getExhaustedRecoveryNextRunAt(payload, now),
+          nextRunAt: getScheduleNextRunAt(payload, now),
           failedCount: sql`COALESCE(${workflowSchedule.failedCount}, 0) + 1`,
           status: sql`CASE WHEN COALESCE(${workflowSchedule.failedCount}, 0) + 1 >= ${MAX_CONSECUTIVE_FAILURES} THEN 'disabled' ELSE 'active' END`,
           infraRetryCount: 0,
@@ -488,8 +498,6 @@ function isStaleDatabaseScheduleJob(job: { status: string; startedAt?: Date }):
 }
 
 async function getDatabaseScheduleExecutionSlots(): Promise<number> {
-  await recoverStaleDatabaseScheduleJobs(new Date())
-
   const [row] = await db
     .select({
       count: sql<number>`count(*)`,
@@ -501,8 +509,6 @@ async function getDatabaseScheduleExecutionSlots(): Promise<number> {
   return Math.max(0, SCHEDULE_EXECUTION_CONCURRENCY_LIMIT - processingCount)
 }
 
-type DatabaseScheduleStartResult = 'started' | 'capacity_full' | 'not_pending'
-
 async function tryStartDatabaseScheduleJob(jobId: string): Promise<DatabaseScheduleStartResult> {
   const now = new Date()
 
@@ -1062,6 +1068,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       let databaseScheduleSlots = SCHEDULE_EXECUTION_CONCURRENCY_LIMIT
 
       if (useDatabaseFallback) {
+        await recoverStaleDatabaseScheduleJobs(queuedAt)
         databaseScheduleSlots = await getDatabaseScheduleExecutionSlots()
         resumedPendingSchedules = await resumePendingDatabaseScheduleJobs(
           jobQueue,

apps/sim/app/api/workspaces/[id]/environment/route.ts

@@ -19,7 +19,10 @@ import {
   createWorkspaceEnvCredentials,
   deleteWorkspaceEnvCredentials,
 } from '@/lib/credentials/environment'
-import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
+import {
+  getPersonalAndWorkspaceEnv,
+  invalidateEffectiveDecryptedEnvCache,
+} from '@/lib/environment/utils'
 import { getUserEntityPermissions, getWorkspaceById } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('WorkspaceEnvironmentAPI')
@@ -136,6 +139,7 @@ export const PUT = withRouteHandler(
         return { existingEncrypted: existing, merged: mergedVars }
       })
 
+      invalidateEffectiveDecryptedEnvCache({ workspaceId })
       const newKeys = Object.keys(variables).filter((k) => !(k in existingEncrypted))
       await createWorkspaceEnvCredentials({ workspaceId, newKeys, actingUserId: userId })
 
@@ -224,6 +228,7 @@ export const DELETE = withRouteHandler(
         return NextResponse.json({ success: true })
       }
 
+      invalidateEffectiveDecryptedEnvCache({ workspaceId })
       await deleteWorkspaceEnvCredentials({ workspaceId, removedKeys: keys })
 
       recordAudit({

apps/sim/lib/copilot/chat/payload.test.ts

@@ -57,11 +57,12 @@ vi.mock('@/tools/params', () => ({
   createUserToolSchema: mockCreateUserToolSchema,
 }))
 
-import { buildIntegrationToolSchemas } from './payload'
+import { buildIntegrationToolSchemas, clearIntegrationToolSchemaCacheForTests } from './payload'
 
 describe('buildIntegrationToolSchemas', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    clearIntegrationToolSchemaCacheForTests()
     mockCreateUserToolSchema.mockReturnValue({ type: 'object', properties: {} })
   })
 
@@ -122,4 +123,16 @@ describe('buildIntegrationToolSchemas', () => {
       { surface: 'copilot' }
     )
   })
+
+  it('briefly reuses built schemas for the same user and surface', async () => {
+    mockGetHighestPrioritySubscription.mockResolvedValue({ plan: 'pro', status: 'active' })
+
+    const first = await buildIntegrationToolSchemas('user-cache')
+    first[0].input_schema.mutated = true
+    const second = await buildIntegrationToolSchemas('user-cache')
+
+    expect(mockGetHighestPrioritySubscription).toHaveBeenCalledTimes(1)
+    expect(mockCreateUserToolSchema).toHaveBeenCalledTimes(3)
+    expect(second[0].input_schema).not.toHaveProperty('mutated')
+  })
 })

apps/sim/lib/copilot/chat/payload.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
+import { LRUCache } from 'lru-cache'
 import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import { isPaid } from '@/lib/billing/plan-helpers'
 import { getToolEntry } from '@/lib/copilot/tool-executor/router'
@@ -11,6 +12,8 @@ import { tools } from '@/tools/registry'
 import { getLatestVersionTools, stripVersionSuffix } from '@/tools/utils'
 
 const logger = createLogger('CopilotChatPayload')
+const INTEGRATION_TOOL_SCHEMA_CACHE_TTL_MS = 5_000
+const INTEGRATION_TOOL_SCHEMA_CACHE_MAX_ENTRIES = 500
 
 interface BuildPayloadParams {
   message: string
@@ -48,6 +51,39 @@ interface BuildIntegrationToolSchemasOptions {
   schemaSurface?: 'default' | 'copilot'
 }
 
+interface IntegrationToolSchemaCacheEntry {
+  promise: Promise<ToolSchema[]>
+}
+
+const integrationToolSchemaCache = new LRUCache<string, IntegrationToolSchemaCacheEntry>({
+  max: INTEGRATION_TOOL_SCHEMA_CACHE_MAX_ENTRIES,
+  ttl: INTEGRATION_TOOL_SCHEMA_CACHE_TTL_MS,
+})
+
+function getIntegrationToolSchemaCacheKey(
+  userId: string,
+  workspaceId: string | undefined,
+  schemaSurface: string
+): string {
+  return JSON.stringify([userId, workspaceId ?? null, schemaSurface])
+}
+
+function cloneToolSchemas(toolSchemas: ToolSchema[]): ToolSchema[] {
+  return toolSchemas.map((tool) => {
+    const cloned: ToolSchema = {
+      ...tool,
+      input_schema: { ...tool.input_schema },
+    }
+    if (tool.params) cloned.params = { ...tool.params }
+    if (tool.oauth) cloned.oauth = { ...tool.oauth }
+    return cloned
+  })
+}
+
+export function clearIntegrationToolSchemaCacheForTests(): void {
+  integrationToolSchemaCache.clear()
+}
+
 /**
  * Build deferred integration tool schemas from the Sim tool registry.
  * Shared by the interactive chat payload builder and the non-interactive
@@ -62,6 +98,36 @@ export async function buildIntegrationToolSchemas(
   messageId?: string,
   options: BuildIntegrationToolSchemasOptions = { schemaSurface: 'copilot' },
   workspaceId?: string
+): Promise<ToolSchema[]> {
+  const schemaSurface = options.schemaSurface ?? 'copilot'
+  const cacheKey = getIntegrationToolSchemaCacheKey(userId, workspaceId, schemaSurface)
+  const cached = integrationToolSchemaCache.get(cacheKey)
+  if (cached) {
+    return cloneToolSchemas(await cached.promise)
+  }
+
+  const promise = buildIntegrationToolSchemasUncached(
+    userId,
+    messageId,
+    { schemaSurface },
+    workspaceId
+  ).catch((error) => {
+    integrationToolSchemaCache.delete(cacheKey)
+    throw error
+  })
+
+  integrationToolSchemaCache.set(cacheKey, {
+    promise,
+  })
+
+  return cloneToolSchemas(await promise)
+}
+
+async function buildIntegrationToolSchemasUncached(
+  userId: string,
+  messageId: string | undefined,
+  options: Required<BuildIntegrationToolSchemasOptions>,
+  workspaceId?: string
 ): Promise<ToolSchema[]> {
   const reqLogger = logger.withMetadata({ messageId })
   const integrationTools: ToolSchema[] = []
@@ -121,7 +187,7 @@ export async function buildIntegrationToolSchemas(
           }
         }
         const userSchema = createUserToolSchema(toolConfig, {
-          surface: options.schemaSurface ?? 'copilot',
+          surface: options.schemaSurface,
         })
         const catalogEntry = getToolEntry(strippedName)
         integrationTools.push({
@@ -165,7 +231,7 @@ export async function buildIntegrationToolSchemas(
     )
   }
 
-  return integrationTools.map((tool) => ({ ...tool, input_schema: { ...tool.input_schema } }))
+  return integrationTools
 }
 
 /**

apps/sim/lib/environment/utils.ts

@@ -4,6 +4,7 @@ import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { eq, inArray } from 'drizzle-orm'
+import { LRUCache } from 'lru-cache'
 import { decryptSecret, encryptSecret } from '@/lib/core/security/encryption'
 import {
   createWorkspaceEnvCredentials,
@@ -13,6 +14,45 @@ import {
 import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('EnvironmentUtils')
+const EFFECTIVE_DECRYPTED_ENV_CACHE_TTL_MS = 2_000
+const EFFECTIVE_DECRYPTED_ENV_CACHE_MAX_ENTRIES = 1_000
+
+interface EffectiveDecryptedEnvCacheEntry {
+  userId: string
+  workspaceId?: string
+  promise: Promise<Record<string, string>>
+}
+
+const effectiveDecryptedEnvCache = new LRUCache<string, EffectiveDecryptedEnvCacheEntry>({
+  max: EFFECTIVE_DECRYPTED_ENV_CACHE_MAX_ENTRIES,
+  ttl: EFFECTIVE_DECRYPTED_ENV_CACHE_TTL_MS,
+})
+
+function getEffectiveDecryptedEnvCacheKey(userId: string, workspaceId?: string): string {
+  return JSON.stringify([userId, workspaceId ?? null])
+}
+
+function cloneEnvVars(envVars: Record<string, string>): Record<string, string> {
+  return { ...envVars }
+}
+
+export function invalidateEffectiveDecryptedEnvCache(input: {
+  userId?: string
+  workspaceId?: string
+}): void {
+  const { userId, workspaceId } = input
+  if (!userId && !workspaceId) return
+
+  effectiveDecryptedEnvCache.forEach((entry, cacheKey) => {
+    if (userId && entry.userId === userId) {
+      effectiveDecryptedEnvCache.delete(cacheKey)
+      return
+    }
+    if (workspaceId && entry.workspaceId === workspaceId) {
+      effectiveDecryptedEnvCache.delete(cacheKey)
+    }
+  })
+}
 
 /**
  * Get environment variable keys for a user
@@ -255,6 +295,7 @@ export async function upsertPersonalEnvVars(
       set: { variables: finalEncrypted, updatedAt: new Date() },
     })
 
+  invalidateEffectiveDecryptedEnvCache({ userId })
   await syncPersonalEnvCredentialsForUser({
     userId,
     envKeys: Object.keys(finalEncrypted),
@@ -304,6 +345,7 @@ export async function upsertWorkspaceEnvVars(
       set: { variables: merged, updatedAt: new Date() },
     })
 
+  invalidateEffectiveDecryptedEnvCache({ workspaceId })
   const newKeys = Object.keys(newVars).filter((k) => !(k in existingWsEncrypted))
   await createWorkspaceEnvCredentials({ workspaceId, newKeys, actingUserId })
 
@@ -317,12 +359,27 @@ export async function getEffectiveDecryptedEnv(
   userId: string,
   workspaceId?: string
 ): Promise<Record<string, string>> {
-  const { personalDecrypted, workspaceDecrypted } = await getPersonalAndWorkspaceEnv(
-    userId,
-    workspaceId
-  )
-  return {
-    ...personalDecrypted,
-    ...workspaceDecrypted,
+  const cacheKey = getEffectiveDecryptedEnvCacheKey(userId, workspaceId)
+  const cached = effectiveDecryptedEnvCache.get(cacheKey)
+  if (cached) {
+    return cloneEnvVars(await cached.promise)
   }
+
+  const promise = getPersonalAndWorkspaceEnv(userId, workspaceId)
+    .then(({ personalDecrypted, workspaceDecrypted }) => ({
+      ...personalDecrypted,
+      ...workspaceDecrypted,
+    }))
+    .catch((error) => {
+      effectiveDecryptedEnvCache.delete(cacheKey)
+      throw error
+    })
+
+  effectiveDecryptedEnvCache.set(cacheKey, {
+    userId,
+    workspaceId,
+    promise,
+  })
+
+  return cloneEnvVars(await promise)
 }

apps/sim/package.json

@@ -149,6 +149,7 @@
     "json5": "2.2.3",
     "jszip": "3.10.1",
     "jwt-decode": "^4.0.0",
+    "lru-cache": "11.5.1",
     "lucide-react": "^0.479.0",
     "mammoth": "^1.9.0",
     "mermaid": "11.15.0",