diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index 5bc20313e08..74462eb6c72 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -432,6 +432,7 @@ export const env = createEnv({ IVM_MAX_OWNER_WEIGHT: z.string().optional().default('5'), // Max accepted weight for weighted owner scheduling IVM_DISTRIBUTED_MAX_INFLIGHT_PER_OWNER:z.string().optional().default('2200'), // Max owner in-flight leases across replicas IVM_DISTRIBUTED_LEASE_MIN_TTL_MS: z.string().optional().default('120000'), // Min TTL for distributed in-flight leases (ms) + IVM_LEASE_REDIS_DEADLINE_MS: z.string().optional().default('1000'), // Deadline for one distributed lease round trip (ms) IVM_QUEUE_TIMEOUT_MS: z.string().optional().default('300000'), // Max queue wait before rejection (ms) IVM_MAX_EXECUTIONS_PER_WORKER: z.string().optional().default('200'), // Max lifetime executions before worker is recycled IVM_MAX_BROKER_ARGS_JSON_CHARS: z.string().optional().default('262144'), // Max JSON payload size for sandbox task broker args (isolate→host) diff --git a/apps/sim/lib/execution/isolated-vm.test.ts b/apps/sim/lib/execution/isolated-vm.test.ts index 9959a369f93..2d77e438c8f 100644 --- a/apps/sim/lib/execution/isolated-vm.test.ts +++ b/apps/sim/lib/execution/isolated-vm.test.ts @@ -8,6 +8,7 @@ import { loggerMock, redisConfigMockFns, } from '@sim/testing' +import { sleep } from '@sim/utils/helpers' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' type MockProc = EventEmitter & { @@ -184,6 +185,7 @@ const { mockSpawn, mockExecSync, mockEnv } = vi.hoisted(() => ({ IVM_MAX_OWNER_WEIGHT: '5', IVM_DISTRIBUTED_MAX_INFLIGHT_PER_OWNER: '100', IVM_DISTRIBUTED_LEASE_MIN_TTL_MS: '1000', + IVM_LEASE_REDIS_DEADLINE_MS: '1000', IVM_QUEUE_TIMEOUT_MS: '1000', IVM_MAX_FETCH_RESPONSE_BYTES: '', IVM_MAX_FETCH_RESPONSE_CHARS: '', @@ -245,6 +247,7 @@ async function loadExecutionModule(options: { IVM_MAX_OWNER_WEIGHT: '5', IVM_DISTRIBUTED_MAX_INFLIGHT_PER_OWNER: '100', IVM_DISTRIBUTED_LEASE_MIN_TTL_MS: '1000', + IVM_LEASE_REDIS_DEADLINE_MS: '1000', IVM_QUEUE_TIMEOUT_MS: '1000', IVM_MAX_FETCH_RESPONSE_BYTES: '', IVM_MAX_FETCH_RESPONSE_CHARS: '', @@ -496,9 +499,10 @@ describe('isolated-vm scheduler', () => { }) expect(result.error?.message).toContain('Too many concurrent') + expect(result.result).toBeNull() }) - it('fails closed when Redis is configured but unavailable', async () => { + it('falls back to local limits when no Redis client is available', async () => { const { executeInIsolatedVM } = await loadExecutionModule({ envOverrides: { REDIS_URL: 'redis://localhost:6379', @@ -516,14 +520,11 @@ describe('isolated-vm scheduler', () => { ownerKey: 'user:redis-down', }) - expect(result.error).toMatchObject({ - isSystemError: true, - message: 'Code execution coordination is temporarily unavailable. Please try again later.', - }) - expect(result.result).toBeNull() + expect(result.error).toBeUndefined() + expect(result.result).toBe('ok') }) - it('fails closed when Redis lease evaluation errors', async () => { + it('falls back to local limits when the lease evaluation errors', async () => { const { executeInIsolatedVM } = await loadExecutionModule({ envOverrides: { REDIS_URL: 'redis://localhost:6379', @@ -548,11 +549,109 @@ describe('isolated-vm scheduler', () => { ownerKey: 'user:redis-error', }) - expect(result.error).toMatchObject({ - isSystemError: true, - message: 'Code execution coordination is temporarily unavailable. Please try again later.', + expect(result.error).toBeUndefined() + expect(result.result).toBe('ok') + }) + + it('falls back to local limits when the lease round trip exceeds its deadline', async () => { + const { executeInIsolatedVM } = await loadExecutionModule({ + envOverrides: { + REDIS_URL: 'redis://localhost:6379', + IVM_LEASE_REDIS_DEADLINE_MS: '5', + }, + spawns: [() => createReadyProc('ok')], + redisEvalImpl: (...args: unknown[]) => { + const script = String(args[0] ?? '') + // Never settles, so only the deadline can decide the outcome. + if (script.includes('ZREMRANGEBYSCORE')) { + return new Promise(() => {}) + } + return 1 + }, + }) + + const result = await executeInIsolatedVM({ + code: 'return "ok"', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-9', + ownerKey: 'user:redis-slow', }) - expect(result.result).toBeNull() + + expect(result.error).toBeUndefined() + expect(result.result).toBe('ok') + }) + + it('releases a lease that Redis registers after the local deadline', async () => { + const scripts: string[] = [] + let completeAcquire!: (value: number) => void + const lateAcquire = new Promise((resolve) => { + completeAcquire = resolve + }) + const { executeInIsolatedVM } = await loadExecutionModule({ + envOverrides: { + REDIS_URL: 'redis://localhost:6379', + IVM_LEASE_REDIS_DEADLINE_MS: '5', + }, + spawns: [() => createReadyProc('ok')], + redisEvalImpl: (...args: unknown[]) => { + const script = String(args[0] ?? '') + scripts.push(script) + // Settles only once the test says so, standing in for a script the + // deadline abandoned locally but that Redis still runs to completion. + if (script.includes('ZREMRANGEBYSCORE')) return lateAcquire + return 1 + }, + }) + + const result = await executeInIsolatedVM({ + code: 'return "ok"', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-11', + ownerKey: 'user:redis-late', + }) + completeAcquire(1) + + expect(result.error).toBeUndefined() + expect(scripts.some((script) => script.includes("'ZREM'"))).toBe(true) + }) + + it('ignores a non-positive configured deadline instead of abandoning every lease', async () => { + const { executeInIsolatedVM } = await loadExecutionModule({ + envOverrides: { + IVM_DISTRIBUTED_MAX_INFLIGHT_PER_OWNER: '1', + IVM_LEASE_REDIS_DEADLINE_MS: '-1', + REDIS_URL: 'redis://localhost:6379', + }, + spawns: [() => createReadyProc('ok')], + redisEvalImpl: async (...args: unknown[]) => { + const script = String(args[0] ?? '') + if (script.includes('ZREMRANGEBYSCORE')) { + // Arrives after a non-positive timer would already have fired, so the + // answer only lands in time when the default deadline is restored. + await sleep(25) + return 0 + } + return 1 + }, + }) + + const result = await executeInIsolatedVM({ + code: 'return "ok"', + params: {}, + envVars: {}, + contextVariables: {}, + timeoutMs: 100, + requestId: 'req-12', + ownerKey: 'user:negative-deadline', + }) + + expect(result.error?.message).toContain('Too many concurrent') }) it('reports cancellation when abort races a rejected distributed lease', async () => { diff --git a/apps/sim/lib/execution/isolated-vm.ts b/apps/sim/lib/execution/isolated-vm.ts index 3081506f75b..4918380f77c 100644 --- a/apps/sim/lib/execution/isolated-vm.ts +++ b/apps/sim/lib/execution/isolated-vm.ts @@ -140,7 +140,22 @@ const MAX_EXECUTIONS_PER_WORKER = Number.parseInt(env.IVM_MAX_EXECUTIONS_PER_WOR const MAX_BROKER_ARGS_JSON_CHARS = Number.parseInt(env.IVM_MAX_BROKER_ARGS_JSON_CHARS) || 262_144 const MAX_BROKERS_PER_EXECUTION = Number.parseInt(env.IVM_MAX_BROKERS_PER_EXECUTION) || 1000 const DISTRIBUTED_KEY_PREFIX = 'ivm:fair:v1:owner' -const LEASE_REDIS_DEADLINE_MS = 200 +/** + * Deadline for a single lease round trip, kept below the shared Redis client's + * `commandTimeout` so this race still resolves first. + * + * Both this deadline and `commandTimeout` are plain `setTimeout`s, so what they + * actually measure is event-loop scheduling, not Redis. A value near normal loop + * latency therefore reports a healthy Redis as unreachable whenever a garbage + * collection pause lands on the call. Keep it well clear of that floor. + * + * A non-positive configured value is treated as unconfigured rather than + * honored: a timer of zero or less fires immediately, which would leave every + * acquisition undetermined and silently drop cross-replica enforcement. + */ +const CONFIGURED_LEASE_REDIS_DEADLINE_MS = Number.parseInt(env.IVM_LEASE_REDIS_DEADLINE_MS) +const LEASE_REDIS_DEADLINE_MS = + CONFIGURED_LEASE_REDIS_DEADLINE_MS > 0 ? CONFIGURED_LEASE_REDIS_DEADLINE_MS : 1000 const QUEUE_RETRY_DELAY_MS = 1000 const DISTRIBUTED_LEASE_GRACE_MS = 30000 @@ -347,7 +362,16 @@ function ownerRedisKey(ownerKey: string): string { return `${DISTRIBUTED_KEY_PREFIX}:${ownerKey}` } -type LeaseAcquireResult = 'acquired' | 'limit_exceeded' | 'unavailable' +/** + * Outcome of one distributed lease acquisition. + * + * `limit_exceeded` is an answer from Redis — the owner is genuinely over its + * share — and is the only outcome that denies an execution. `undetermined` + * means no answer arrived before the deadline, which is not a denial and must + * never be projected as one: the local admission limits below still bound the + * work, so the caller falls back to them. + */ +type LeaseAcquireResult = 'acquired' | 'limit_exceeded' | 'undetermined' async function tryAcquireDistributedLease( ownerKey: string, @@ -358,10 +382,10 @@ async function tryAcquireDistributedLease( const redis = getRedisClient() if (!redis) { - logger.error('Redis is configured but unavailable for distributed lease acquisition', { + logger.warn('No Redis client for distributed lease acquisition; using local limits', { ownerKey, }) - return 'unavailable' + return 'undetermined' } const now = Date.now() @@ -407,11 +431,12 @@ async function tryAcquireDistributedLease( ]) return Number(result) === 1 ? 'acquired' : 'limit_exceeded' } catch (error) { - logger.error('Failed to acquire distributed owner lease; execution will be rejected', { + logger.warn('Distributed owner lease undetermined; using local limits', { ownerKey, + deadlineMs: LEASE_REDIS_DEADLINE_MS, error, }) - return 'unavailable' + return 'undetermined' } finally { clearTimeout(deadlineTimer) } @@ -1416,23 +1441,18 @@ export async function executeInIsolatedVM( }, } } - if (leaseAcquireResult === 'unavailable') { - logger.error('Isolated-vm execution rejected because its distributed lease is unavailable', { - ownerKey, - }) - maybeCleanupOwner(ownerKey) - return { - result: null, - stdout: '', - error: { - message: 'Code execution coordination is temporarily unavailable. Please try again later.', - name: 'Error', - isSystemError: true, - }, - } - } + // An undetermined lease cannot reject the execution: the per-process pool and + // the per-owner active/queued limits above still bound this work. let settled = false + /** + * Released even when the acquisition was undetermined. The deadline abandons + * the local wait but cannot cancel the script, so a late completion still + * registers this lease id — and unreleased it would count against the owner + * for the whole TTL, denying later executions that do have capacity. The + * lease id is unique to this execution, so removing one that was never + * registered is a no-op. + */ const releaseLease = () => { if (settled) return settled = true