houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Slow instance commands (#19431)

Browse Source 
# Introduction
This PR introduces the slow instance commands pattern, that allow
migrating data in prior of the schema migration, that would fail if not.

Slow instance commands runs after the fast instance commands and before
the workspace commands.
On twenty instance that do not has any active or suspended workspace the
data migration part is skipped but the migration still runs, especially
for fresh installs

We were previously hacking through typeorm transaction system to gain
such granularity using save points:
```ts
export class AddPayloadToCommandMenuItem1775129635528
  implements MigrationInterface
{
  name = 'AddPayloadToCommandMenuItem1775129635528';

  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(
      `ALTER TABLE "core"."commandMenuItem" ADD "payload" jsonb`,
    );

    const savepointName =
      'sp_add_payload_check_constraint_to_command_menu_item';

    try {
      await queryRunner.query(`SAVEPOINT ${savepointName}`);

      await addPayloadCheckConstraintToCommandMenuItem(queryRunner);

      await queryRunner.query(`RELEASE SAVEPOINT ${savepointName}`);
    } catch (e) {
      try {
        await queryRunner.query(`ROLLBACK TO SAVEPOINT ${savepointName}`);
        await queryRunner.query(`RELEASE SAVEPOINT ${savepointName}`);
      } catch (rollbackError) {
        // oxlint-disable-next-line no-console
        console.error(
          'Failed to rollback to savepoint in AddPayloadToCommandMenuItem1775129635528',
          rollbackError,
        );
        throw rollbackError;
      }

      // oxlint-disable-next-line no-console
      console.error(
        'Swallowing AddPayloadToCommandMenuItem1775129635528 error',
        e,
      );
    }
  }
```

It was afterwards re-applied within an workspace commands, it was hacky
and missleading for the self host having false positive in logs

## New pattern

Generate the slow instance command
```
npx nx database:migrate:generate twenty-server -- --name add-foo-bar-columns --type slow
```

```ts
import { DataSource, QueryRunner } from 'typeorm';

import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator';
import { SlowInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/slow-instance-command.interface';

@RegisteredInstanceCommand('1.21.0', 1775640902366, { type: 'slow' })
export class AddPrastoinColToWorkspaceSlowInstanceCommand implements SlowInstanceCommand {
  async runDataMigration(dataSource: DataSource): Promise<void> {
    // TODO: implement data backfill before the DDL migration
  }

  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query('ALTER TABLE "core"."workspace" ADD "prastoin" character varying');
  }

  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query('ALTER TABLE "core"."workspace" DROP COLUMN "prastoin"');
  }
}
```

### Why `run-instance-commands` and `upgrade` remain separate commands

These two commands serve fundamentally different purposes with
incompatible scoping semantics:
The run-instance-commands iterates over all the legacy typeorm and the
instance commands of all versions, used for database init and so on. we
could be centralizing both but the readability tradeoff isn't worth it

In the future thanks to the cross-version pattern we will be able to
centralize them but not right now

Please note that by default the `run-instance-commands` only run the
fast instance commands which is expected for our cloud prod CD
This commit is contained in:
Paul Rastoin
2026-04-08 14:09:29 +02:00
committed by GitHub
parent 8a84e32cf6
commit 85be463487
13 changed files with 873 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/database/commands/upgrade-version-command/__tests__/upgrade.command.spec.ts
+183 -95
View File
@@ -3,11 +3,10 @@ import {
eachTestingContextFilter,
type EachTestingContext,
} from 'twenty-shared/testing';
import {
type DataSource,
type MigrationInterface,
type QueryRunner,
} from 'typeorm';
import { type DataSource, type QueryRunner } from 'typeorm';
import { type FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface';
import { type SlowInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/slow-instance-command.interface';
import { getDataSourceToken } from '@nestjs/typeorm';
@@ -61,7 +60,7 @@ type BuildUpgradeCommandModuleArgs = {
workspaces: WorkspaceEntity[];
appVersion: string | null;
commandRunner: CommandRunnerValues;
migrations?: MigrationInterface[];
migrations?: FastInstanceCommand[];
};
const buildUpgradeCommandModule = async ({
workspaces,
@@ -92,8 +91,11 @@ const buildUpgradeCommandModule = async ({
: {
provide: UpgradeCommandRegistryService,
useValue: {
getInstanceCommandsForVersion: jest.fn().mockReturnValue([]),
getWorkspaceCommandsForVersion: jest.fn().mockReturnValue([]),
getBundleForVersion: jest.fn().mockReturnValue({
fastInstanceCommands: [],
slowInstanceCommands: [],
workspaceCommands: [],
}),
},
};
@@ -184,7 +186,10 @@ const buildUpgradeCommandModule = async ({
{
provide: InstanceUpgradeService,
useValue: {
runSingleMigration: jest
runFastInstanceCommand: jest
.fn()
.mockResolvedValue({ status: 'success' }),
runSlowInstanceCommand: jest
.fn()
.mockResolvedValue({ status: 'success' }),
},
@@ -242,7 +247,7 @@ describe('UpgradeCommandRunner', () => {
workspaces?: WorkspaceEntity[];
appVersion?: string | null;
commandRunner?: CommandRunnerValues;
migrations?: MigrationInterface[];
migrations?: FastInstanceCommand[];
};
const buildModuleAndSetupSpies = async ({
numberOfWorkspace = 1,
@@ -365,26 +370,26 @@ describe('UpgradeCommandRunner', () => {
);
});
it('should call runSingleMigration for each current-version instance command', async () => {
it('should call runFastInstanceCommand for each current-version instance command', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class AddIndexToUsers1770000000000 implements MigrationInterface {
class AddIndexToUsers1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(CURRENT_VERSION, 1771000000000)
class AddColumnToAccounts1771000000000 implements MigrationInterface {
class AddColumnToAccounts1771000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(PREVIOUS_VERSION, 1769000000000)
class DropLegacyTable1769000000000 implements MigrationInterface {
class DropLegacyTable1769000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
class UndecoratedMigration1768000000000 implements MigrationInterface {
class UndecoratedMigration1768000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@@ -400,105 +405,46 @@ describe('UpgradeCommandRunner', () => {
const instanceUpgradeService = module.get(InstanceUpgradeService);
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run([], {});
await upgradeCommandRunner.run(passedParams, options);
expect(instanceUpgradeService.runSingleMigration).toHaveBeenCalledTimes(2);
expect(instanceUpgradeService.runSingleMigration).toHaveBeenNthCalledWith(
1,
addIndex,
);
expect(instanceUpgradeService.runSingleMigration).toHaveBeenNthCalledWith(
expect(instanceUpgradeService.runFastInstanceCommand).toHaveBeenCalledTimes(
2,
addColumn,
);
expect(
instanceUpgradeService.runFastInstanceCommand,
).toHaveBeenNthCalledWith(1, {
command: addIndex,
name: `${CURRENT_VERSION}_AddIndexToUsers1770000000000_1770000000000`,
});
expect(
instanceUpgradeService.runFastInstanceCommand,
).toHaveBeenNthCalledWith(2, {
command: addColumn,
name: `${CURRENT_VERSION}_AddColumnToAccounts1771000000000_1771000000000`,
});
});
it('should skip already-executed instance commands', async () => {
it('should propagate errors from runFastInstanceCommand', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class AlreadyRunMigration1770000000000 implements MigrationInterface {
class FailingMigration1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const alreadyRun = new AlreadyRunMigration1770000000000();
const module = await buildModuleAndSetupSpies({
migrations: [alreadyRun],
migrations: [new FailingMigration1770000000000()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(instanceUpgradeService.runSingleMigration as jest.Mock).mockResolvedValue({
status: 'already-executed',
});
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run(passedParams, options);
expect(upgradeCommandRunner['logger'].warn).toHaveBeenCalledWith(
expect.stringContaining('already executed'),
);
});
it('should throw when a migration fails', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class FailingMigration1770000000000 implements MigrationInterface {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const failing = new FailingMigration1770000000000();
const module = await buildModuleAndSetupSpies({
migrations: [failing],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(instanceUpgradeService.runSingleMigration as jest.Mock).mockResolvedValue({
(
instanceUpgradeService.runFastInstanceCommand as jest.Mock
).mockResolvedValue({
status: 'failed',
error: new Error('SQL error'),
});
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await expect(
upgradeCommandRunner.run(passedParams, options),
).rejects.toThrow('Core migration FailingMigration1770000000000 failed');
});
it('should log success when a migration succeeds', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class SuccessMigration1770000000000 implements MigrationInterface {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const success = new SuccessMigration1770000000000();
const module = await buildModuleAndSetupSpies({
migrations: [success],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
const passedParams: string[] = [];
const options: UpgradeCommandOptions = {};
await upgradeCommandRunner.run(passedParams, options);
expect(instanceUpgradeService.runSingleMigration).toHaveBeenCalledWith(
success,
);
expect(upgradeCommandRunner['logger'].log).toHaveBeenCalledWith(
expect.stringContaining('executed successfully'),
);
await expect(upgradeCommandRunner.run([], {})).rejects.toThrow('SQL error');
});
describe('Workspace upgrade should fail', () => {
@@ -565,4 +511,146 @@ describe('UpgradeCommandRunner', () => {
},
);
});
it('should call runSlowInstanceCommand for each current-version slow command', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigration1780000000000 implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const slowMigration = new SlowMigration1780000000000();
const module = await buildModuleAndSetupSpies({
migrations: [slowMigration],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
await upgradeCommandRunner.run([], {});
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledTimes(
1,
);
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledWith({
command: slowMigration,
name: `${CURRENT_VERSION}_SlowMigration1780000000000_1780000000000`,
skipDataMigration: false,
});
});
it('should run slow commands after fast commands but before workspace commands', async () => {
const executionOrder: string[] = [];
@RegisteredInstanceCommand(CURRENT_VERSION, 1770000000000)
class FastMigration1770000000000 implements FastInstanceCommand {
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigration1780000000000 implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
migrations: [
new FastMigration1770000000000(),
new SlowMigration1780000000000(),
],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(
instanceUpgradeService.runFastInstanceCommand as jest.Mock
).mockImplementation(async () => {
executionOrder.push('fast');
return { status: 'success' };
});
(
instanceUpgradeService.runSlowInstanceCommand as jest.Mock
).mockImplementation(async () => {
executionOrder.push('slow');
return { status: 'success' };
});
const workspaceIteratorService = module.get(WorkspaceIteratorService);
(workspaceIteratorService.iterate as jest.Mock).mockImplementation(
async () => {
executionOrder.push('workspace');
return { success: [], fail: [] };
},
);
await upgradeCommandRunner.run([], {});
expect(executionOrder).toStrictEqual(['fast', 'slow', 'workspace']);
});
it('should pass skipDataMigration: true on fresh install (no workspaces)', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class SlowMigrationFreshInstall implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
numberOfWorkspace: 0,
migrations: [new SlowMigrationFreshInstall()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
await upgradeCommandRunner.run([], {});
expect(instanceUpgradeService.runSlowInstanceCommand).toHaveBeenCalledWith({
command: expect.any(SlowMigrationFreshInstall),
name: `${CURRENT_VERSION}_SlowMigrationFreshInstall_1780000000000`,
skipDataMigration: true,
});
});
it('should propagate errors from runSlowInstanceCommand', async () => {
@RegisteredInstanceCommand(CURRENT_VERSION, 1780000000000, {
type: 'slow',
})
class FailingSlowMigration implements SlowInstanceCommand {
async runDataMigration(_dataSource: DataSource): Promise<void> {}
async up(_queryRunner: QueryRunner) {}
async down(_queryRunner: QueryRunner) {}
}
const module = await buildModuleAndSetupSpies({
migrations: [new FailingSlowMigration()],
});
const instanceUpgradeService = module.get(InstanceUpgradeService);
(
instanceUpgradeService.runSlowInstanceCommand as jest.Mock
).mockResolvedValue({
status: 'failed',
error: new Error('Data migration error'),
});
await expect(upgradeCommandRunner.run([], {})).rejects.toThrow(
'Data migration error',
);
});
});
packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts
+42 -60
View File
@@ -3,8 +3,7 @@ import { InjectDataSource } from '@nestjs/typeorm';
import chalk from 'chalk';
import { Command, CommandRunner, Option } from 'nest-commander';
import { SemVer } from 'semver';
import { assertUnreachable, isDefined } from 'twenty-shared/utils';
import { DataSource, MigrationInterface } from 'typeorm';
import { DataSource } from 'typeorm';
import { ActiveOrSuspendedWorkspaceCommandRunner } from 'src/database/commands/command-runners/active-or-suspended-workspace.command-runner';
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
@@ -13,7 +12,10 @@ import { CommandLogger } from 'src/database/commands/logger';
import { type UpgradeCommandVersion } from 'src/engine/constants/upgrade-command-supported-versions.constant';
import { CoreEngineVersionService } from 'src/engine/core-engine-version/services/core-engine-version.service';
import { InstanceUpgradeService } from 'src/engine/core-modules/upgrade/services/instance-upgrade.service';
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import {
UpgradeCommandRegistryService,
type VersionBundle,
} from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
import { WorkspaceUpgradeService } from 'src/engine/core-modules/upgrade/services/workspace-upgrade.service';
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
@@ -30,12 +32,10 @@ export type UpgradeCommandOptions = {
verbose?: boolean;
};
type VersionContext = {
type VersionContext = VersionBundle & {
fromWorkspaceVersion: SemVer;
currentAppVersion: SemVer;
currentVersionMajorMinor: UpgradeCommandVersion;
instanceCommands: MigrationInterface[];
workspaceCommands: VersionCommands;
};
@Command({
@@ -143,7 +143,8 @@ export class UpgradeCommand extends CommandRunner {
'Initialized upgrade context with:',
`- currentVersion (migrating to): ${versionContext.currentAppVersion}`,
`- fromWorkspaceVersion: ${versionContext.fromWorkspaceVersion}`,
`- ${versionContext.instanceCommands.length} instance commands (from registry)`,
`- ${versionContext.fastInstanceCommands.length} fast instance commands (from registry)`,
`- ${versionContext.slowInstanceCommands.length} slow instance commands (from registry)`,
`- ${versionContext.workspaceCommands.length} workspace commands`,
].join('\n '),
),
@@ -167,11 +168,37 @@ Please roll back to that version and run the upgrade command again.`,
}
await this.runLegacyPendingTypeOrmMigrations();
await this.runInstanceCommandsOrThrow(versionContext);
for (const { command, name } of versionContext.fastInstanceCommands) {
const result = await this.instanceUpgradeService.runFastInstanceCommand(
{
command,
name,
},
);
if (result.status === 'failed') {
throw result.error;
}
}
const hasWorkspaces =
await this.workspaceVersionService.hasActiveOrSuspendedWorkspaces();
for (const { command, name } of versionContext.slowInstanceCommands) {
const result = await this.instanceUpgradeService.runSlowInstanceCommand(
{
command,
name,
skipDataMigration: !hasWorkspaces,
},
);
if (result.status === 'failed') {
throw result.error;
}
}
if (!hasWorkspaces) {
this.logger.log(
chalk.blue(
@@ -220,64 +247,16 @@ Please roll back to that version and run the upgrade command again.`,
}
}
private async runInstanceCommandsOrThrow(
versionContext: VersionContext,
): Promise<void> {
for (const instanceCommand of versionContext.instanceCommands) {
const migrationName = instanceCommand.constructor.name;
const result =
await this.instanceUpgradeService.runSingleMigration(instanceCommand);
switch (result.status) {
case 'already-executed': {
this.logger.warn(
`Core migration ${migrationName} already executed, skipping`,
);
break;
}
case 'failed': {
this.logger.error(`Core migration ${migrationName} failed`);
if (isDefined(result.error)) {
this.logger.error(
result.error instanceof Error
? (result.error.stack ?? result.error.message)
: String(result.error),
);
}
throw new Error(`Core migration ${migrationName} failed`);
}
case 'success': {
this.logger.log(
`Core migration ${migrationName} executed successfully`,
);
break;
}
default: {
assertUnreachable(result);
}
}
}
}
private resolveVersionContext(): VersionContext {
const currentAppVersion = this.coreEngineVersionService.getCurrentVersion();
const currentVersionMajorMinor =
`${currentAppVersion.major}.${currentAppVersion.minor}.0` as UpgradeCommandVersion;
const workspaceCommands =
this.upgradeCommandRegistryService.getWorkspaceCommandsForVersion(
currentVersionMajorMinor,
);
const fromWorkspaceVersion =
this.coreEngineVersionService.getPreviousVersion();
const instanceCommands =
this.upgradeCommandRegistryService.getInstanceCommandsForVersion(
const { fastInstanceCommands, slowInstanceCommands, workspaceCommands } =
this.upgradeCommandRegistryService.getBundleForVersion(
currentVersionMajorMinor,
);
@@ -285,8 +264,9 @@ Please roll back to that version and run the upgrade command again.`,
fromWorkspaceVersion,
currentAppVersion,
currentVersionMajorMinor,
fastInstanceCommands,
slowInstanceCommands,
workspaceCommands,
instanceCommands,
};
}
@@ -308,7 +288,9 @@ Please roll back to that version and run the upgrade command again.`,
options,
fromWorkspaceVersion: versionContext.fromWorkspaceVersion,
currentAppVersion: versionContext.currentAppVersion,
workspaceCommands: versionContext.workspaceCommands,
workspaceCommands: versionContext.workspaceCommands.map(
(entry) => entry.command,
),
});
},
});