diff --git a/packages/twenty-server/jest-integration.config.ts b/packages/twenty-server/jest-integration.config.ts index ddfa4d9062..ce5968f6bb 100644 --- a/packages/twenty-server/jest-integration.config.ts +++ b/packages/twenty-server/jest-integration.config.ts @@ -35,6 +35,7 @@ const jestConfig: JestConfigWithTsJest = { modulePathIgnorePatterns: ['/dist'], globalSetup: '/test/integration/utils/setup-test.ts', globalTeardown: '/test/integration/utils/teardown-test.ts', + setupFilesAfterEnv: ['/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts'], testTimeout: 20000, maxWorkers: 1, // jsdom 29 pulls ESM-only transitive deps (parse5, entities, tough-cookie, @@ -81,9 +82,6 @@ const jestConfig: JestConfigWithTsJest = { }), '^test/(.*)$': '/test/$1', }, - fakeTimers: { - enableGlobally: true, - }, globals: { APP_PORT: 4000, NODE_ENV: NodeEnvironment.TEST, diff --git a/packages/twenty-server/test/integration/graphql/suites/auth/jwt-key-rotation/rotate-signing-keys-cron-job.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/auth/jwt-key-rotation/rotate-signing-keys-cron-job.integration-spec.ts index 6fac2d7688..7fc775fea8 100644 --- a/packages/twenty-server/test/integration/graphql/suites/auth/jwt-key-rotation/rotate-signing-keys-cron-job.integration-spec.ts +++ b/packages/twenty-server/test/integration/graphql/suites/auth/jwt-key-rotation/rotate-signing-keys-cron-job.integration-spec.ts @@ -6,6 +6,7 @@ import { findManyApplications } from 'test/integration/graphql/utils/find-many-a import { generateApiKeyToken } from 'test/integration/graphql/utils/generate-api-key-token.util'; import { deleteConfigVariable } from 'test/integration/twenty-config/utils/delete-config-variable.util'; import { updateConfigVariable } from 'test/integration/twenty-config/utils/update-config-variable.util'; +import { expectEventually } from 'test/integration/utils/expect-eventually.util'; import { RotateSigningKeysCronJob } from 'src/engine/core-modules/jwt/crons/jobs/rotate-signing-keys.cron.job'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; @@ -14,7 +15,6 @@ import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-q import { API_KEY_DATA_SEED_IDS } from 'src/engine/workspace-manager/dev-seeder/data/constants/api-key-data-seeds.constant'; const SIGNING_KEY_ROTATION_DAYS_KEY = 'SIGNING_KEY_ROTATION_DAYS'; -const ROTATE_SIGNING_KEYS_CRON_PATTERN = '15 3 * * *'; describe('RotateSigningKeysCronJob (integration)', () => { const seededApiKeyId = API_KEY_DATA_SEED_IDS.ID_1; @@ -62,30 +62,30 @@ describe('RotateSigningKeysCronJob (integration)', () => { input: { key: SIGNING_KEY_ROTATION_DAYS_KEY, value: 0 }, }); - await cronQueue.addCron({ - jobName: RotateSigningKeysCronJob.name, - data: undefined, - options: { repeat: { pattern: ROTATE_SIGNING_KEYS_CRON_PATTERN } }, + await cronQueue.add(RotateSigningKeysCronJob.name, {}); + + let rotatedApiKeyToken = ''; + + await expectEventually(async () => { + const rotatedTokenResponse = await generateApiKeyToken({ + apiKeyId: seededApiKeyId, + accessToken: APPLE_JANE_ADMIN_ACCESS_TOKEN, + }); + + expect(rotatedTokenResponse.body.errors).toBeUndefined(); + + rotatedApiKeyToken = + rotatedTokenResponse.body.data?.generateApiKeyToken.token ?? ''; + + expect(isNonEmptyString(rotatedApiKeyToken)).toBe(true); + + const rotatedKid = decodeJwtCompleteOrThrow(rotatedApiKeyToken).header + .kid as string; + + expect(isNonEmptyString(rotatedKid)).toBe(true); + expect(rotatedKid).not.toBe(initialKid); }); - const rotatedTokenResponse = await generateApiKeyToken({ - apiKeyId: seededApiKeyId, - accessToken: APPLE_JANE_ADMIN_ACCESS_TOKEN, - }); - - expect(rotatedTokenResponse.body.errors).toBeUndefined(); - - const rotatedApiKeyToken: string = - rotatedTokenResponse.body.data?.generateApiKeyToken.token ?? ''; - - expect(isNonEmptyString(rotatedApiKeyToken)).toBe(true); - - const rotatedKid = decodeJwtCompleteOrThrow(rotatedApiKeyToken).header - .kid as string; - - expect(isNonEmptyString(rotatedKid)).toBe(true); - expect(rotatedKid).not.toBe(initialKid); - const callWithPreviousToken = await findManyApplications({ accessToken: initialApiKeyToken, expectToFail: false, diff --git a/packages/twenty-server/test/integration/graphql/suites/nested-relation-per-parent-limit.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/nested-relation-per-parent-limit.integration-spec.ts index 1c4f8b7331..6db2eaf445 100644 --- a/packages/twenty-server/test/integration/graphql/suites/nested-relation-per-parent-limit.integration-spec.ts +++ b/packages/twenty-server/test/integration/graphql/suites/nested-relation-per-parent-limit.integration-spec.ts @@ -59,7 +59,7 @@ describe('Nested relation per-parent limit (e2e)', () => { { id: SOFT_DELETED_COMPANY_ID, name: 'Soft-deleted relation company' }, { id: EMPTY_COMPANY_ID, name: 'Empty relation company' }, ], - upsert: true, + upsert: false, }); await makeGraphqlAPIRequest(createCompanies); @@ -76,7 +76,7 @@ describe('Nested relation per-parent limit (e2e)', () => { companyId: SOFT_DELETED_COMPANY_ID, })), ], - upsert: true, + upsert: false, }); await makeGraphqlAPIRequest(createPeople); diff --git a/packages/twenty-server/test/integration/graphql/suites/search/search-resolver.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/search/search-resolver.integration-spec.ts index b481db79b4..e62adbd653 100644 --- a/packages/twenty-server/test/integration/graphql/suites/search/search-resolver.integration-spec.ts +++ b/packages/twenty-server/test/integration/graphql/suites/search/search-resolver.integration-spec.ts @@ -26,6 +26,7 @@ import { import { createManyOperation } from 'test/integration/graphql/utils/create-many-operation.util'; import { search } from 'test/integration/graphql/utils/search.util'; import { deleteAllRecords } from 'test/integration/utils/delete-all-records'; +import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util'; import { eachTestingContextFilter, type EachTestingContext, @@ -192,6 +193,9 @@ describe('SearchResolver', () => { data: persons, }); + await waitForAllJobsToFinish(); + await deleteAllRecords('company'); + await createManyOperation({ objectMetadataSingularName: 'company', objectMetadataPluralName: 'companies', diff --git a/packages/twenty-server/test/integration/graphql/suites/workflow/quick-lead-workflow.integration-spec.ts b/packages/twenty-server/test/integration/graphql/suites/workflow/quick-lead-workflow.integration-spec.ts index 114b6cc750..4a1bb1fec4 100644 --- a/packages/twenty-server/test/integration/graphql/suites/workflow/quick-lead-workflow.integration-spec.ts +++ b/packages/twenty-server/test/integration/graphql/suites/workflow/quick-lead-workflow.integration-spec.ts @@ -3,6 +3,8 @@ import { destroyWorkflowRun, getWorkflowRun, runWorkflowVersion, + waitForWorkflowCompletion, + waitForWorkflowRunStatus, } from 'test/integration/graphql/suites/workflow/utils/workflow-run-test.util'; import { v4 as uuidv4 } from 'uuid'; @@ -125,7 +127,10 @@ describe('Quick Lead Workflow (e2e)', () => { createdWorkflowRunId = workflowRunId; - const workflowRun = await getWorkflowRun(workflowRunId); + const workflowRun = await waitForWorkflowRunStatus( + workflowRunId, + 'RUNNING', + ); expect(workflowRun).toBeDefined(); expect(workflowRun?.workflowVersionId).toBe( @@ -238,7 +243,10 @@ describe('Quick Lead Workflow (e2e)', () => { expect(testWorkflowRunId).toBeDefined(); - let workflowRun = await getWorkflowRun(testWorkflowRunId as string); + let workflowRun = await waitForWorkflowRunStatus( + testWorkflowRunId as string, + 'RUNNING', + ); expect(workflowRun?.status).toBe('RUNNING'); expect(workflowRun?.state?.stepInfos?.[FORM_STEP_ID]?.status).toBe( @@ -276,7 +284,9 @@ describe('Quick Lead Workflow (e2e)', () => { expect(submitFormResponse.body.errors).toBeUndefined(); expect(submitFormResponse.body.data.submitFormStep).toBe(true); - workflowRun = await getWorkflowRun(testWorkflowRunId as string); + workflowRun = await waitForWorkflowCompletion( + testWorkflowRunId as string, + ); expect(workflowRun?.status).toBe('COMPLETED'); expect(workflowRun?.state?.stepInfos?.trigger?.status).toBe('SUCCESS'); expect(workflowRun?.state?.stepInfos?.[FORM_STEP_ID]?.status).toBe( diff --git a/packages/twenty-server/test/integration/graphql/suites/workflow/utils/workflow-run-test.util.ts b/packages/twenty-server/test/integration/graphql/suites/workflow/utils/workflow-run-test.util.ts index b8fa4efa76..385a23a6f4 100644 --- a/packages/twenty-server/test/integration/graphql/suites/workflow/utils/workflow-run-test.util.ts +++ b/packages/twenty-server/test/integration/graphql/suites/workflow/utils/workflow-run-test.util.ts @@ -117,6 +117,13 @@ export const destroyWorkflowRun = async ( }); }; +const PENDING_WORKFLOW_RUN_STATUSES: WorkflowRunStatusType[] = [ + 'NOT_STARTED', + 'ENQUEUED', + 'RUNNING', + 'STOPPING', +]; + export const waitForWorkflowCompletion = async ( workflowRunId: string, maxAttempts = 30, @@ -126,9 +133,32 @@ export const waitForWorkflowCompletion = async ( let attempts = 0; while ( - workflowRun?.status === 'RUNNING' && - attempts < maxAttempts && - workflowRun !== null + workflowRun !== null && + PENDING_WORKFLOW_RUN_STATUSES.includes(workflowRun.status) && + attempts < maxAttempts + ) { + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + workflowRun = await getWorkflowRun(workflowRunId); + attempts++; + } + + return workflowRun; +}; + +export const waitForWorkflowRunStatus = async ( + workflowRunId: string, + expectedStatus: WorkflowRunStatusType, + maxAttempts = 30, + intervalMs = 500, +): Promise => { + let workflowRun = await getWorkflowRun(workflowRunId); + let attempts = 0; + + while ( + attempts < maxAttempts && + workflowRun?.status !== expectedStatus && + (workflowRun === null || + PENDING_WORKFLOW_RUN_STATUSES.includes(workflowRun.status)) ) { await new Promise((resolve) => setTimeout(resolve, intervalMs)); workflowRun = await getWorkflowRun(workflowRunId); diff --git a/packages/twenty-server/test/integration/utils/create-app.ts b/packages/twenty-server/test/integration/utils/create-app.ts index 37355fa586..eabf12acc6 100644 --- a/packages/twenty-server/test/integration/utils/create-app.ts +++ b/packages/twenty-server/test/integration/utils/create-app.ts @@ -17,9 +17,7 @@ import { CaptchaDriverFactory } from 'src/engine/core-modules/captcha/captcha-dr import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { ExceptionHandlerMockService } from 'src/engine/core-modules/exception-handler/mocks/exception-handler-mock.service'; import { MockedUnhandledExceptionFilter } from 'src/engine/core-modules/exception-handler/mocks/mock-unhandled-exception.filter'; -import { SyncDriver } from 'src/engine/core-modules/message-queue/drivers/sync.driver'; import { JobsModule } from 'src/engine/core-modules/message-queue/jobs.module'; -import { QUEUE_DRIVER } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueModule } from 'src/engine/core-modules/message-queue/message-queue.module'; interface TestingModuleCreatePreHook { @@ -33,10 +31,6 @@ export type TestingAppCreatePreHook = ( app: NestExpressApplication, ) => Promise; -// Shared SyncDriver instance for all queues in tests -// This enables synchronous processing of jobs during integration tests -const syncDriver = new SyncDriver(); - /** * Sets basic integration testing module of app */ @@ -66,9 +60,7 @@ export const createApp = async ( getCurrentDriver: () => ({ validate: async () => ({ success: true }), }), - }) - .overrideProvider(QUEUE_DRIVER) - .useValue(syncDriver); + }); if (config.moduleBuilderHook) { moduleBuilder = config.moduleBuilderHook(moduleBuilder); diff --git a/packages/twenty-server/test/integration/utils/expect-eventually.util.ts b/packages/twenty-server/test/integration/utils/expect-eventually.util.ts new file mode 100644 index 0000000000..4efcb1ccb8 --- /dev/null +++ b/packages/twenty-server/test/integration/utils/expect-eventually.util.ts @@ -0,0 +1,25 @@ +type ExpectEventuallyOptions = { + timeoutMs?: number; + intervalMs?: number; +}; + +export const expectEventually = async ( + assertion: () => Promise | void, + { timeoutMs = 10_000, intervalMs = 100 }: ExpectEventuallyOptions = {}, +): Promise => { + const startedAt = Date.now(); + + for (;;) { + try { + await assertion(); + + return; + } catch (error) { + if (Date.now() - startedAt > timeoutMs) { + throw error; + } + + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + } +}; diff --git a/packages/twenty-server/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts b/packages/twenty-server/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts new file mode 100644 index 0000000000..3d4ae90d3d --- /dev/null +++ b/packages/twenty-server/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts @@ -0,0 +1,19 @@ +import { + closeQueueConnections, + waitForAllJobsToFinish, +} from 'test/integration/utils/wait-for-all-jobs-to-finish.util'; + +const WAIT_FOR_JOBS_HOOK_TIMEOUT_MS = 150_000; + +beforeAll(async () => { + await waitForAllJobsToFinish(); +}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS); + +afterEach(async () => { + jest.useRealTimers(); + await waitForAllJobsToFinish(); +}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS); + +afterAll(async () => { + await closeQueueConnections(); +}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS); diff --git a/packages/twenty-server/test/integration/utils/wait-for-all-jobs-to-finish.util.ts b/packages/twenty-server/test/integration/utils/wait-for-all-jobs-to-finish.util.ts new file mode 100644 index 0000000000..e53f07c593 --- /dev/null +++ b/packages/twenty-server/test/integration/utils/wait-for-all-jobs-to-finish.util.ts @@ -0,0 +1,131 @@ +import { Queue } from 'bullmq'; +import IORedis from 'ioredis'; + +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; + +const POLL_INTERVAL_MS = 25; +const REQUIRED_CONSECUTIVE_QUIET_CHECKS = 2; +const STALL_TIMEOUT_MS = 15_000; +const HARD_TIMEOUT_MS = 120_000; +const PENDING_JOB_STATES = [ + 'waiting', + 'active', + 'prioritized', + 'waiting-children', + 'delayed', +] as const; + +let redisConnection: IORedis | null = null; +let queues: Queue[] | null = null; + +const getQueues = (): Queue[] => { + if (!queues) { + redisConnection = new IORedis( + process.env.REDIS_QUEUE_URL ?? + process.env.REDIS_URL ?? + 'redis://localhost:6379', + { maxRetriesPerRequest: null }, + ); + queues = Object.values(MessageQueue).map( + (queueName) => new Queue(queueName, { connection: redisConnection! }), + ); + } + + return queues; +}; + +const getPendingJobCountsByQueue = async (): Promise< + Record +> => { + const countsByQueue = await Promise.all( + getQueues().map(async (queue) => { + const jobCounts = await queue.getJobCounts(...PENDING_JOB_STATES); + const pendingCount = Object.values(jobCounts).reduce( + (sum, count) => sum + count, + 0, + ); + + return [queue.name, pendingCount] as const; + }), + ); + + return Object.fromEntries( + countsByQueue.filter(([, pendingCount]) => pendingCount > 0), + ); +}; + +const getActiveJobsFingerprint = async ( + busyQueueNames: string[], +): Promise => { + const activeJobIdsByQueue = await Promise.all( + getQueues() + .filter((queue) => busyQueueNames.includes(queue.name)) + .map(async (queue) => { + const activeJobs = await queue.getActive(0, 50); + + return `${queue.name}:${activeJobs.map((job) => job.id).join(',')}`; + }), + ); + + return activeJobIdsByQueue.join('|'); +}; + +export const waitForAllJobsToFinish = async (): Promise => { + const startedAt = Date.now(); + let lastProgressAt = startedAt; + let lastBusyFingerprint = ''; + let consecutiveQuietChecks = 0; + + while (consecutiveQuietChecks < REQUIRED_CONSECUTIVE_QUIET_CHECKS) { + const pendingJobCountsByQueue = await getPendingJobCountsByQueue(); + const pendingTotal = Object.values(pendingJobCountsByQueue).reduce( + (sum, count) => sum + count, + 0, + ); + + if (pendingTotal === 0) { + consecutiveQuietChecks += 1; + } else { + consecutiveQuietChecks = 0; + const now = Date.now(); + + const activeJobsFingerprint = await getActiveJobsFingerprint( + Object.keys(pendingJobCountsByQueue), + ); + const busyFingerprint = `${pendingTotal}|${activeJobsFingerprint}`; + + if (busyFingerprint !== lastBusyFingerprint) { + lastBusyFingerprint = busyFingerprint; + lastProgressAt = now; + } + + if (now - lastProgressAt > STALL_TIMEOUT_MS) { + throw new Error( + `Message queues stalled, no progress for ${STALL_TIMEOUT_MS}ms, pending jobs: ${JSON.stringify(pendingJobCountsByQueue)}`, + ); + } + + if (now - startedAt > HARD_TIMEOUT_MS) { + throw new Error( + `Message queues still busy after ${HARD_TIMEOUT_MS}ms, pending jobs: ${JSON.stringify(pendingJobCountsByQueue)}`, + ); + } + } + + if (consecutiveQuietChecks < REQUIRED_CONSECUTIVE_QUIET_CHECKS) { + await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS)); + } + } +}; + +export const closeQueueConnections = async (): Promise => { + if (queues) { + await Promise.allSettled(queues.map((queue) => queue.close())); + queues = null; + } + + if (redisConnection) { + await redisConnection.quit().catch(() => redisConnection?.disconnect()); + redisConnection = null; + } +};