[Upgrade] Fix workspace creation cursor (#19701)
## Summary ### Problem The upgrade migration system required new workspaces to always start from a workspace command, which was too rigid. When the system was mid-upgrade within an instance command (IC) segment, workspace creation would fail or produce inconsistent state. ### Solution #### Workspace-scoped instance command rows Instance commands now write upgrade migration rows for **all active/suspended workspaces** alongside the global row. This means every workspace has a complete migration history, including instance command records. - `InstanceCommandRunnerService` reloads `activeOrSuspendedWorkspaceIds` immediately before writing records (both success and failure paths) to mitigate race conditions with concurrent workspace creation. - `recordUpgradeMigration` in `UpgradeMigrationService` accepts a discriminated union over `status`, handles `error: unknown` formatting internally, and writes global + workspace rows in batch. #### Flexible initial cursor for new workspaces `getInitialCursorForNewWorkspace` now accepts the last **attempted** (not just completed) instance command with its status: - If the IC is `completed` and the next step is a workspace segment → cursor is set to the last WC of that segment (existing behavior). - If the IC is `failed` or not the last of its segment → cursor is set to that IC itself, preserving its status. This allows workspaces to be created at any point during the upgrade lifecycle, including mid-IC-segment and after IC failure. #### Relaxed workspace segment validation `validateWorkspaceCursorsAreInWorkspaceSegment` accepts workspaces whose cursor is: 1. Within the current workspace segment, OR 2. At the immediately preceding instance command with `completed` status (handles the `-w` single-workspace upgrade scenario). Workspaces with cursors in a previous segment, ahead of the current segment, or at a preceding IC with `failed` status are rejected. ### Test plan created empty workspaces to allow testing upgrade with several active workspaces
This commit is contained in:
Paul Rastoin
-3
@@ -1,3 +0,0 @@
|
||||
// Jest Snapshot v1, https://jestjs.io/docs/snapshot-testing
|
||||
|
||||
exports[`UpgradeSequenceRunnerService — failing sequence (integration) should throw when cursor command is not found in the sequence 1`] = `"Step "RemovedCommand" not found in upgrade sequence. The sequence only covers versions [1.21.0, 1.22.0, 1.23.0]. Please upgrade to 1.21.0 first."`;
|
||||
-369
@@ -1,369 +0,0 @@
|
||||
import {
|
||||
type UpgradeStep,
|
||||
type WorkspaceUpgradeStep,
|
||||
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
|
||||
import {
|
||||
type IntegrationTestContext,
|
||||
createUpgradeSequenceRunnerIntegrationTestModule,
|
||||
DEFAULT_OPTIONS,
|
||||
makeFastInstance,
|
||||
makeSlowInstance,
|
||||
makeStep,
|
||||
makeWorkspace,
|
||||
resetSeedSequenceCounter,
|
||||
seedMigration,
|
||||
setMockActiveWorkspaceIds,
|
||||
testGetLatestMigrationForCommand,
|
||||
WS_1,
|
||||
WS_2,
|
||||
} from './utils/upgrade-sequence-runner-integration-test.util';
|
||||
|
||||
const makeFailingFastInstance = (name: string, error: Error): UpgradeStep =>
|
||||
({
|
||||
...makeStep('fast-instance', name),
|
||||
command: {
|
||||
up: async () => {
|
||||
throw error;
|
||||
},
|
||||
down: async () => {},
|
||||
},
|
||||
}) as unknown as UpgradeStep;
|
||||
|
||||
const makeFailingWorkspace = (
|
||||
name: string,
|
||||
error: Error,
|
||||
): WorkspaceUpgradeStep =>
|
||||
({
|
||||
...makeStep('workspace', name),
|
||||
command: {
|
||||
runOnWorkspace: async () => {
|
||||
throw error;
|
||||
},
|
||||
},
|
||||
}) as unknown as WorkspaceUpgradeStep;
|
||||
|
||||
const makeWorkspaceFailingForIds = (
|
||||
name: string,
|
||||
failingWorkspaceIds: Set<string>,
|
||||
error: Error,
|
||||
): WorkspaceUpgradeStep =>
|
||||
({
|
||||
...makeStep('workspace', name),
|
||||
command: {
|
||||
runOnWorkspace: async ({ workspaceId }: { workspaceId: string }) => {
|
||||
if (failingWorkspaceIds.has(workspaceId)) {
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
},
|
||||
}) as unknown as WorkspaceUpgradeStep;
|
||||
|
||||
describe('UpgradeSequenceRunnerService — failing sequence (integration)', () => {
|
||||
let context: IntegrationTestContext;
|
||||
|
||||
beforeAll(async () => {
|
||||
context = await createUpgradeSequenceRunnerIntegrationTestModule();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await context.dataSource.query('DELETE FROM core."upgradeMigration"');
|
||||
await context.module?.close();
|
||||
await context.dataSource?.destroy();
|
||||
}, 15000);
|
||||
|
||||
beforeEach(async () => {
|
||||
await context.dataSource.query('DELETE FROM core."upgradeMigration"');
|
||||
resetSeedSequenceCounter();
|
||||
setMockActiveWorkspaceIds([]);
|
||||
jest.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('should throw when no migration history exists', async () => {
|
||||
const sequence = [makeFastInstance('Ic1')];
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrow(
|
||||
'No upgrade migration found — the database may not have been initialized',
|
||||
);
|
||||
});
|
||||
|
||||
it('should throw when cursor command is not found in the sequence', async () => {
|
||||
const sequence = [makeFastInstance('Ic1'), makeFastInstance('Ic2')];
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'RemovedCommand',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrowErrorMatchingSnapshot();
|
||||
});
|
||||
|
||||
it('should throw when workspace cursors are outside the current slice', async () => {
|
||||
const sequence = [
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc3'),
|
||||
makeWorkspace('Wc4'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1, WS_2]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc3',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc4',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrow('workspaces are not aligned');
|
||||
});
|
||||
|
||||
it('should throw when an active workspace has no migration history', async () => {
|
||||
const sequence = [makeFastInstance('Ic1'), makeWorkspace('Wc1')];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1, WS_2]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrow('No upgrade migration found for workspace(s)');
|
||||
});
|
||||
|
||||
it('should record failure in DB when a fast instance command fails', async () => {
|
||||
const error = new Error('fast command exploded');
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeFailingFastInstance('Ic2', error),
|
||||
];
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrow('fast command exploded');
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'failed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should record failure in DB when a slow instance command fails', async () => {
|
||||
const error = new Error('slow data migration exploded');
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
{
|
||||
...makeSlowInstance('Ic2'),
|
||||
command: {
|
||||
up: async () => {},
|
||||
down: async () => {},
|
||||
runDataMigration: async () => {
|
||||
throw error;
|
||||
},
|
||||
},
|
||||
} as unknown as UpgradeStep,
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
await expect(
|
||||
context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
}),
|
||||
).rejects.toThrow('slow data migration exploded');
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'failed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should abort and report failures when workspace commands fail, without running subsequent instance steps', async () => {
|
||||
const error = new Error('workspace command exploded');
|
||||
const sequence = [
|
||||
makeFailingWorkspace('Wc1', error),
|
||||
makeFastInstance('Ic1'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'failed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(report.totalFailures).toBe(1);
|
||||
expect(report.totalSuccesses).toBe(0);
|
||||
|
||||
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
});
|
||||
|
||||
expect(ic1).toBeNull();
|
||||
|
||||
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(wc1).toEqual(
|
||||
expect.objectContaining({ status: 'failed', attempt: 2 }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should abort at workspace failure in a multi-segment sequence with two workspaces starting aligned', async () => {
|
||||
const error = new Error('Wc2 exploded for WS_2');
|
||||
|
||||
const sequence = [
|
||||
makeWorkspace('Wc0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspaceFailingForIds('Wc2', new Set([WS_2]), error),
|
||||
makeFastInstance('Ic2'),
|
||||
makeWorkspace('Wc3'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1, WS_2]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc0',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc0',
|
||||
status: 'completed',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: {
|
||||
...DEFAULT_OPTIONS,
|
||||
workspaceIds: [WS_1, WS_2],
|
||||
},
|
||||
});
|
||||
|
||||
expect(report.totalSuccesses).toBe(1);
|
||||
expect(report.totalFailures).toBe(1);
|
||||
|
||||
// WS_1 succeeded Wc2
|
||||
const ws1Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(ws1Wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
|
||||
// WS_2 failed Wc2
|
||||
const ws2Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
|
||||
expect(ws2Wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'failed' }),
|
||||
);
|
||||
|
||||
// Ic2 never ran — runner aborted at the workspace segment failure
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
|
||||
expect(ic2).toBeNull();
|
||||
|
||||
// Wc3 never ran either
|
||||
const ws1Wc3 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc3',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(ws1Wc3).toBeNull();
|
||||
});
|
||||
});
|
||||
-466
@@ -1,466 +0,0 @@
|
||||
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
|
||||
|
||||
import {
|
||||
type IntegrationTestContext,
|
||||
createUpgradeSequenceRunnerIntegrationTestModule,
|
||||
DEFAULT_OPTIONS,
|
||||
makeFastInstance,
|
||||
makeSlowInstance,
|
||||
makeWorkspace,
|
||||
resetSeedSequenceCounter,
|
||||
seedMigration,
|
||||
setMockActiveWorkspaceIds,
|
||||
testGetLatestMigrationForCommand,
|
||||
WS_1,
|
||||
WS_2,
|
||||
} from './utils/upgrade-sequence-runner-integration-test.util';
|
||||
|
||||
describe('UpgradeSequenceRunnerService — execution (integration)', () => {
|
||||
let context: IntegrationTestContext;
|
||||
|
||||
beforeAll(async () => {
|
||||
context = await createUpgradeSequenceRunnerIntegrationTestModule();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await context.dataSource.query('DELETE FROM core."upgradeMigration"');
|
||||
await context.module?.close();
|
||||
await context.dataSource?.destroy();
|
||||
}, 15000);
|
||||
|
||||
beforeEach(async () => {
|
||||
await context.dataSource.query('DELETE FROM core."upgradeMigration"');
|
||||
resetSeedSequenceCounter();
|
||||
setMockActiveWorkspaceIds([]);
|
||||
jest.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('should return zero counts for an empty sequence', async () => {
|
||||
const report = await context.runner.run({
|
||||
sequence: [],
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(report).toEqual({ totalSuccesses: 0, totalFailures: 0 });
|
||||
});
|
||||
|
||||
it('should resume from a completed instance command and run remaining steps', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeFastInstance('Ic2'),
|
||||
makeSlowInstance('Ic3'),
|
||||
];
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
});
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
const ic3 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic3',
|
||||
});
|
||||
|
||||
expect(ic1).toEqual(
|
||||
expect.objectContaining({ status: 'completed', attempt: 1 }),
|
||||
);
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ status: 'completed', attempt: 1 }),
|
||||
);
|
||||
expect(ic3).toEqual(expect.objectContaining({ status: 'completed' }));
|
||||
});
|
||||
|
||||
it('should retry a failed instance command', async () => {
|
||||
const sequence = [makeFastInstance('Ic1'), makeFastInstance('Ic2')];
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
status: 'failed',
|
||||
});
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'completed', attempt: 2 }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should resume workspace commands from per-workspace cursors', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should enforce workspace sync barrier before instance step', async () => {
|
||||
const sequence = [makeWorkspace('Wc1'), makeFastInstance('Ic1')];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
});
|
||||
|
||||
expect(ic1).toEqual(
|
||||
expect.objectContaining({ name: 'Ic1', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should skip data migration for slow instance commands when no workspaces exist', async () => {
|
||||
const sequence = [makeFastInstance('Ic1'), makeSlowInstance('Ic2')];
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
const instanceCommandRunnerService = context.module.get(
|
||||
InstanceCommandRunnerService,
|
||||
);
|
||||
const spy = jest.spyOn(
|
||||
instanceCommandRunnerService,
|
||||
'runSlowInstanceCommand',
|
||||
);
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(spy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ skipDataMigration: true }),
|
||||
);
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should run data migration for slow instance commands when workspaces exist', async () => {
|
||||
const sequence = [makeFastInstance('Ic0'), makeSlowInstance('Ic1')];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
const instanceCommandRunnerService = context.module.get(
|
||||
InstanceCommandRunnerService,
|
||||
);
|
||||
const spy = jest.spyOn(
|
||||
instanceCommandRunnerService,
|
||||
'runSlowInstanceCommand',
|
||||
);
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(spy).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ skipDataMigration: false }),
|
||||
);
|
||||
|
||||
const ic1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
});
|
||||
|
||||
expect(ic1).toEqual(
|
||||
expect.objectContaining({ name: 'Ic1', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should run workspace commands for multiple workspaces successfully', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1, WS_2]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: {
|
||||
...DEFAULT_OPTIONS,
|
||||
workspaceIds: [WS_1, WS_2],
|
||||
},
|
||||
});
|
||||
|
||||
expect(report).toEqual({ totalSuccesses: 2, totalFailures: 0 });
|
||||
|
||||
const ws1Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
const ws2Wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
|
||||
expect(ws1Wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
expect(ws2Wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should execute the full sequence from the initial cursor on a fresh run', async () => {
|
||||
const sequence = [
|
||||
makeWorkspace('Wc0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeFastInstance('Ic2'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc0',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
|
||||
);
|
||||
expect(wc1).toEqual(
|
||||
expect.objectContaining({ name: 'Wc1', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should retry a failed workspace command', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'failed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(report.totalFailures).toBe(0);
|
||||
|
||||
const wc1 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(wc1).toEqual(
|
||||
expect.objectContaining({
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
attempt: 2,
|
||||
}),
|
||||
);
|
||||
expect(wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should traverse a multi-segment sequence with sync barriers', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeFastInstance('Ic2'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(report.totalFailures).toBe(0);
|
||||
|
||||
const ic2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Ic2',
|
||||
});
|
||||
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(ic2).toEqual(
|
||||
expect.objectContaining({ name: 'Ic2', status: 'completed' }),
|
||||
);
|
||||
expect(wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('should ignore migration records from inactive workspaces when resolving the global cursor', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
setMockActiveWorkspaceIds([WS_1]);
|
||||
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc1',
|
||||
status: 'completed',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
// WS_2 is inactive — its record is more recent (seeded later)
|
||||
// but should not influence the global cursor
|
||||
await seedMigration(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
status: 'completed',
|
||||
workspaceId: WS_2,
|
||||
});
|
||||
|
||||
const report = await context.runner.run({
|
||||
sequence,
|
||||
options: DEFAULT_OPTIONS,
|
||||
});
|
||||
|
||||
expect(report.totalFailures).toBe(0);
|
||||
|
||||
const wc2 = await testGetLatestMigrationForCommand(context.dataSource, {
|
||||
name: 'Wc2',
|
||||
workspaceId: WS_1,
|
||||
});
|
||||
|
||||
expect(wc2).toEqual(
|
||||
expect.objectContaining({ name: 'Wc2', status: 'completed' }),
|
||||
);
|
||||
});
|
||||
});
|
||||
+299
@@ -0,0 +1,299 @@
|
||||
import 'reflect-metadata';
|
||||
|
||||
import { Test } from '@nestjs/testing';
|
||||
import { DiscoveryService } from '@nestjs/core';
|
||||
|
||||
import {
|
||||
type UpgradeStep,
|
||||
UpgradeSequenceReaderService,
|
||||
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
|
||||
import { UpgradeCommandRegistryService } from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
|
||||
import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator';
|
||||
import { TWENTY_CURRENT_VERSION } from 'src/engine/core-modules/upgrade/constants/twenty-current-version.constant';
|
||||
|
||||
const VERSION = TWENTY_CURRENT_VERSION;
|
||||
|
||||
@RegisteredWorkspaceCommand(VERSION, 1770000000000)
|
||||
class MinimalWorkspaceCommand {
|
||||
async runOnWorkspace(): Promise<void> {}
|
||||
}
|
||||
|
||||
const buildProviderWrapper = (instance: object) => ({
|
||||
instance,
|
||||
metatype: instance.constructor,
|
||||
});
|
||||
|
||||
const buildServiceWithMockedSequence = async (
|
||||
mockSequence: UpgradeStep[],
|
||||
): Promise<UpgradeSequenceReaderService> => {
|
||||
const module = await Test.createTestingModule({
|
||||
providers: [
|
||||
UpgradeSequenceReaderService,
|
||||
UpgradeCommandRegistryService,
|
||||
{
|
||||
provide: DiscoveryService,
|
||||
useValue: {
|
||||
getProviders: () =>
|
||||
[new MinimalWorkspaceCommand()].map(buildProviderWrapper),
|
||||
},
|
||||
},
|
||||
],
|
||||
}).compile();
|
||||
|
||||
const registryService = module.get(UpgradeCommandRegistryService);
|
||||
|
||||
registryService.onModuleInit();
|
||||
|
||||
const service = module.get(UpgradeSequenceReaderService);
|
||||
|
||||
jest.spyOn(service, 'getUpgradeSequence').mockReturnValue(mockSequence);
|
||||
|
||||
return service;
|
||||
};
|
||||
|
||||
const noopAsync = async () => {};
|
||||
|
||||
const makeStep = (kind: UpgradeStep['kind'], name: string): UpgradeStep => {
|
||||
const command =
|
||||
kind === 'workspace'
|
||||
? { runOnWorkspace: noopAsync }
|
||||
: kind === 'slow-instance'
|
||||
? { up: noopAsync, down: noopAsync, runDataMigration: noopAsync }
|
||||
: { up: noopAsync, down: noopAsync };
|
||||
|
||||
return {
|
||||
kind,
|
||||
name,
|
||||
command,
|
||||
version: VERSION,
|
||||
timestamp: 0,
|
||||
} as unknown as UpgradeStep;
|
||||
};
|
||||
|
||||
const makeFastInstance = (name: string) => makeStep('fast-instance', name);
|
||||
const makeWorkspace = (name: string) => makeStep('workspace', name);
|
||||
|
||||
describe('UpgradeSequenceReaderService', () => {
|
||||
describe('getInitialCursorForNewWorkspace', () => {
|
||||
it('should return last workspace command of segment following completed instance command', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Wc2', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return the instance command itself when next step is another instance command', async () => {
|
||||
const sequence = [
|
||||
makeWorkspace('Wc-1'),
|
||||
makeFastInstance('Ic0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic0', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return last workspace command when all instance commands in batch are completed', async () => {
|
||||
const sequence = [
|
||||
makeWorkspace('Wc-1'),
|
||||
makeFastInstance('Ic0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Wc1', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should stop at next instance command boundary', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc2'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Wc1', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return the instance command itself when at end of sequence', async () => {
|
||||
const sequence = [makeWorkspace('Wc0'), makeFastInstance('Ic0')];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic0', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return the instance command itself when no workspace command exists before it', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc0'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic0', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return final segment when last instance command is completed', async () => {
|
||||
const sequence = [
|
||||
makeWorkspace('Wc0'),
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc2'),
|
||||
makeWorkspace('Wc3'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic1',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Wc3', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should handle single workspace command in segment', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Wc0', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return the instance command itself when sequence ends with instance commands batch', async () => {
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeFastInstance('Ic2'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic2',
|
||||
status: 'completed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic2', status: 'completed' });
|
||||
});
|
||||
|
||||
it('should return the failed instance command when IC failed — not skip forward to WC segment', async () => {
|
||||
// Sequence: Ic0 → Ic1 → Wc0 → Wc1
|
||||
// Ic1 failed → cursor stays at Ic1:failed (does NOT skip to Wc1)
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic1',
|
||||
status: 'failed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic1', status: 'failed' });
|
||||
});
|
||||
|
||||
it('should return the failed instance command even when next step is a workspace command', async () => {
|
||||
// Sequence: Ic0 → Wc0 → Wc1
|
||||
// Ic0 failed → cursor stays at Ic0:failed
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeWorkspace('Wc0'),
|
||||
makeWorkspace('Wc1'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic0',
|
||||
status: 'failed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic0', status: 'failed' });
|
||||
});
|
||||
|
||||
it('should return the failed mid-segment instance command', async () => {
|
||||
// Sequence: Ic0 → Ic1 → Ic2 → Wc0
|
||||
// Ic1 failed (Ic0 completed but Ic1 is the last attempted) → cursor at Ic1:failed
|
||||
const sequence = [
|
||||
makeFastInstance('Ic0'),
|
||||
makeFastInstance('Ic1'),
|
||||
makeFastInstance('Ic2'),
|
||||
makeWorkspace('Wc0'),
|
||||
];
|
||||
|
||||
const service = await buildServiceWithMockedSequence(sequence);
|
||||
|
||||
const result = service.getInitialCursorForNewWorkspace({
|
||||
name: 'Ic1',
|
||||
status: 'failed',
|
||||
});
|
||||
|
||||
expect(result).toEqual({ name: 'Ic1', status: 'failed' });
|
||||
});
|
||||
});
|
||||
});
|
||||
-275
@@ -1,275 +0,0 @@
|
||||
import { Test, type TestingModule } from '@nestjs/testing';
|
||||
import { getDataSourceToken, getRepositoryToken } from '@nestjs/typeorm';
|
||||
|
||||
import { config } from 'dotenv';
|
||||
import { DataSource, type Repository } from 'typeorm';
|
||||
|
||||
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
|
||||
import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
|
||||
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
|
||||
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
|
||||
import {
|
||||
UpgradeSequenceReaderService,
|
||||
type UpgradeStep,
|
||||
type WorkspaceUpgradeStep,
|
||||
} from 'src/engine/core-modules/upgrade/services/upgrade-sequence-reader.service';
|
||||
import { UpgradeSequenceRunnerService } from 'src/engine/core-modules/upgrade/services/upgrade-sequence-runner.service';
|
||||
import { WorkspaceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/workspace-command-runner.service';
|
||||
import { UpgradeMigrationEntity } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
|
||||
import {
|
||||
SEED_APPLE_WORKSPACE_ID,
|
||||
SEED_YCOMBINATOR_WORKSPACE_ID,
|
||||
} from 'src/engine/workspace-manager/dev-seeder/core/constants/seeder-workspaces.constant';
|
||||
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
|
||||
|
||||
jest.useRealTimers();
|
||||
|
||||
config({
|
||||
path: process.env.NODE_ENV === 'test' ? '.env.test' : '.env',
|
||||
override: true,
|
||||
});
|
||||
|
||||
export const WS_1 = SEED_APPLE_WORKSPACE_ID;
|
||||
export const WS_2 = SEED_YCOMBINATOR_WORKSPACE_ID;
|
||||
|
||||
const EXECUTED_BY_VERSION = '42.42.42';
|
||||
|
||||
const noopAsync = async () => {};
|
||||
|
||||
export const makeStep = (
|
||||
kind: UpgradeStep['kind'],
|
||||
name: string,
|
||||
): UpgradeStep => {
|
||||
const command =
|
||||
kind === 'workspace'
|
||||
? { runOnWorkspace: noopAsync }
|
||||
: kind === 'slow-instance'
|
||||
? { up: noopAsync, down: noopAsync, runDataMigration: noopAsync }
|
||||
: { up: noopAsync, down: noopAsync };
|
||||
|
||||
return {
|
||||
kind,
|
||||
name,
|
||||
command,
|
||||
version: '1.21.0',
|
||||
timestamp: 0,
|
||||
} as unknown as UpgradeStep;
|
||||
};
|
||||
|
||||
export const makeFastInstance = (name: string) =>
|
||||
makeStep('fast-instance', name);
|
||||
|
||||
export const makeSlowInstance = (name: string) =>
|
||||
makeStep('slow-instance', name);
|
||||
|
||||
export const makeWorkspace = (name: string) =>
|
||||
makeStep('workspace', name) as WorkspaceUpgradeStep;
|
||||
|
||||
let mockActiveWorkspaceIds: string[] = [];
|
||||
|
||||
export const setMockActiveWorkspaceIds = (ids: string[]) => {
|
||||
mockActiveWorkspaceIds = ids;
|
||||
};
|
||||
|
||||
export const DEFAULT_OPTIONS = {
|
||||
workspaceIds: undefined,
|
||||
startFromWorkspaceId: undefined,
|
||||
workspaceCountLimit: undefined,
|
||||
dryRun: false,
|
||||
verbose: false,
|
||||
};
|
||||
|
||||
type IntegrationTestModule = Awaited<
|
||||
ReturnType<typeof createUpgradeSequenceRunnerIntegrationTestModule>
|
||||
>;
|
||||
|
||||
export type IntegrationTestContext = {
|
||||
[K in keyof IntegrationTestModule]: IntegrationTestModule[K];
|
||||
};
|
||||
|
||||
export const createUpgradeSequenceRunnerIntegrationTestModule = async () => {
|
||||
const dataSource = new DataSource({
|
||||
type: 'postgres',
|
||||
url: process.env.PG_DATABASE_URL,
|
||||
schema: 'core',
|
||||
entities: [
|
||||
'src/engine/core-modules/**/*.entity.ts',
|
||||
'src/engine/metadata-modules/**/*.entity.ts',
|
||||
],
|
||||
synchronize: false,
|
||||
});
|
||||
|
||||
await dataSource.initialize();
|
||||
|
||||
const migrationRepo: Repository<UpgradeMigrationEntity> =
|
||||
dataSource.getRepository(UpgradeMigrationEntity);
|
||||
|
||||
const module: TestingModule = await Test.createTestingModule({
|
||||
providers: [
|
||||
{
|
||||
provide: getRepositoryToken(UpgradeMigrationEntity),
|
||||
useValue: migrationRepo,
|
||||
},
|
||||
{
|
||||
provide: getDataSourceToken(),
|
||||
useValue: dataSource,
|
||||
},
|
||||
{
|
||||
provide: TwentyConfigService,
|
||||
useValue: {
|
||||
get: (key: string) =>
|
||||
key === 'APP_VERSION' ? EXECUTED_BY_VERSION : undefined,
|
||||
},
|
||||
},
|
||||
UpgradeMigrationService,
|
||||
{
|
||||
provide: WorkspaceVersionService,
|
||||
useValue: {
|
||||
getActiveOrSuspendedWorkspaceIds: jest
|
||||
.fn()
|
||||
.mockImplementation(async () => mockActiveWorkspaceIds),
|
||||
hasActiveOrSuspendedWorkspaces: jest
|
||||
.fn()
|
||||
.mockImplementation(async () => mockActiveWorkspaceIds.length > 0),
|
||||
},
|
||||
},
|
||||
{
|
||||
provide: UpgradeSequenceReaderService,
|
||||
useFactory: () => new UpgradeSequenceReaderService({} as any),
|
||||
},
|
||||
InstanceCommandRunnerService,
|
||||
WorkspaceCommandRunnerService,
|
||||
{
|
||||
provide: WorkspaceIteratorService,
|
||||
useValue: {
|
||||
iterate: jest.fn().mockImplementation(async (args: any) => {
|
||||
const { callback, workspaceIds } = args;
|
||||
const ids = workspaceIds ?? [WS_1];
|
||||
const report = { fail: [] as any[], success: [] as any[] };
|
||||
|
||||
for (const [index, workspaceId] of ids.entries()) {
|
||||
try {
|
||||
await callback({
|
||||
workspaceId,
|
||||
index,
|
||||
total: ids.length,
|
||||
dataSource,
|
||||
});
|
||||
report.success.push({ workspaceId });
|
||||
} catch (error) {
|
||||
report.fail.push({ error, workspaceId });
|
||||
}
|
||||
}
|
||||
|
||||
return report;
|
||||
}),
|
||||
},
|
||||
},
|
||||
UpgradeSequenceRunnerService,
|
||||
],
|
||||
}).compile();
|
||||
|
||||
const runner = module.get(UpgradeSequenceRunnerService);
|
||||
|
||||
jest.spyOn(runner['logger'], 'log').mockImplementation();
|
||||
jest.spyOn(runner['logger'], 'error').mockImplementation();
|
||||
jest.spyOn(runner['logger'], 'warn').mockImplementation();
|
||||
|
||||
const instanceCommandRunnerService = module.get(InstanceCommandRunnerService);
|
||||
|
||||
jest
|
||||
.spyOn(instanceCommandRunnerService['logger'], 'log')
|
||||
.mockImplementation();
|
||||
jest
|
||||
.spyOn(instanceCommandRunnerService['logger'], 'error')
|
||||
.mockImplementation();
|
||||
|
||||
const workspaceCommandRunnerService = module.get(
|
||||
WorkspaceCommandRunnerService,
|
||||
);
|
||||
|
||||
jest
|
||||
.spyOn(workspaceCommandRunnerService['logger'], 'log')
|
||||
.mockImplementation();
|
||||
jest
|
||||
.spyOn(workspaceCommandRunnerService['logger'], 'error')
|
||||
.mockImplementation();
|
||||
|
||||
return {
|
||||
module,
|
||||
dataSource,
|
||||
runner,
|
||||
};
|
||||
};
|
||||
|
||||
let seedSequenceCounter = 0;
|
||||
|
||||
export const resetSeedSequenceCounter = () => {
|
||||
seedSequenceCounter = 0;
|
||||
};
|
||||
|
||||
export const seedMigration = async (
|
||||
dataSource: DataSource,
|
||||
{
|
||||
name,
|
||||
status,
|
||||
workspaceId = null,
|
||||
attempt = 1,
|
||||
}: {
|
||||
name: string;
|
||||
status: 'completed' | 'failed';
|
||||
workspaceId?: string | null;
|
||||
attempt?: number;
|
||||
},
|
||||
) => {
|
||||
const createdAt = new Date(
|
||||
Date.now() + seedSequenceCounter * 1000,
|
||||
).toISOString();
|
||||
|
||||
seedSequenceCounter++;
|
||||
|
||||
await dataSource.query(
|
||||
`INSERT INTO core."upgradeMigration" (name, status, attempt, "executedByVersion", "workspaceId", "createdAt")
|
||||
VALUES ($1, $2, $3, $4, $5, $6)`,
|
||||
[name, status, attempt, EXECUTED_BY_VERSION, workspaceId, createdAt],
|
||||
);
|
||||
};
|
||||
|
||||
export const testCountMigrationsForCommand = async (
|
||||
dataSource: DataSource,
|
||||
{
|
||||
name,
|
||||
workspaceId = null,
|
||||
}: {
|
||||
name: string;
|
||||
workspaceId?: string | null;
|
||||
},
|
||||
): Promise<number> => {
|
||||
const rows = await dataSource.query(
|
||||
`SELECT COUNT(*)::int AS count FROM core."upgradeMigration"
|
||||
WHERE name = $1 AND ($2::uuid IS NULL AND "workspaceId" IS NULL OR "workspaceId" = $2)`,
|
||||
[name, workspaceId],
|
||||
);
|
||||
|
||||
return rows[0].count;
|
||||
};
|
||||
|
||||
export const testGetLatestMigrationForCommand = async (
|
||||
dataSource: DataSource,
|
||||
{
|
||||
name,
|
||||
workspaceId = null,
|
||||
}: {
|
||||
name: string;
|
||||
workspaceId?: string | null;
|
||||
},
|
||||
): Promise<{ name: string; status: string; attempt: number } | null> => {
|
||||
const rows = await dataSource.query(
|
||||
`SELECT name, status, attempt FROM core."upgradeMigration"
|
||||
WHERE name = $1 AND ($2::uuid IS NULL AND "workspaceId" IS NULL OR "workspaceId" = $2)
|
||||
ORDER BY attempt DESC LIMIT 1`,
|
||||
[name, workspaceId],
|
||||
);
|
||||
|
||||
return rows.length > 0 ? rows[0] : null;
|
||||
};
|
||||
+29
-7
@@ -7,6 +7,7 @@ import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twent
|
||||
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 { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
|
||||
import { WorkspaceVersionService } from 'src/engine/workspace-manager/workspace-version/services/workspace-version.service';
|
||||
|
||||
type RunSingleMigrationResult =
|
||||
| { status: 'success' }
|
||||
@@ -22,6 +23,7 @@ export class InstanceCommandRunnerService {
|
||||
private readonly dataSource: DataSource,
|
||||
private readonly twentyConfigService: TwentyConfigService,
|
||||
private readonly upgradeMigrationService: UpgradeMigrationService,
|
||||
private readonly workspaceVersionService: WorkspaceVersionService,
|
||||
) {}
|
||||
|
||||
async runFastInstanceCommand({
|
||||
@@ -54,9 +56,16 @@ export class InstanceCommandRunnerService {
|
||||
|
||||
await command.up(queryRunner);
|
||||
|
||||
await this.upgradeMigrationService.markAsCompleted({
|
||||
const workspaceIds =
|
||||
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds({
|
||||
queryRunner,
|
||||
});
|
||||
|
||||
await this.upgradeMigrationService.recordUpgradeMigration({
|
||||
name,
|
||||
workspaceId: null,
|
||||
workspaceIds,
|
||||
isInstance: true,
|
||||
status: 'completed',
|
||||
executedByVersion,
|
||||
queryRunner,
|
||||
});
|
||||
@@ -67,9 +76,14 @@ export class InstanceCommandRunnerService {
|
||||
await queryRunner.rollbackTransaction();
|
||||
}
|
||||
|
||||
await this.upgradeMigrationService.markAsFailed({
|
||||
const workspaceIds =
|
||||
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds();
|
||||
|
||||
await this.upgradeMigrationService.recordUpgradeMigration({
|
||||
name,
|
||||
workspaceId: null,
|
||||
workspaceIds,
|
||||
isInstance: true,
|
||||
status: 'failed',
|
||||
executedByVersion,
|
||||
error,
|
||||
});
|
||||
@@ -117,9 +131,14 @@ export class InstanceCommandRunnerService {
|
||||
try {
|
||||
await command.runDataMigration(this.dataSource);
|
||||
} catch (error) {
|
||||
await this.upgradeMigrationService.markAsFailed({
|
||||
const workspaceIds =
|
||||
await this.workspaceVersionService.getActiveOrSuspendedWorkspaceIds();
|
||||
|
||||
await this.upgradeMigrationService.recordUpgradeMigration({
|
||||
name,
|
||||
workspaceId: null,
|
||||
workspaceIds,
|
||||
isInstance: true,
|
||||
status: 'failed',
|
||||
executedByVersion,
|
||||
error,
|
||||
});
|
||||
@@ -133,6 +152,9 @@ export class InstanceCommandRunnerService {
|
||||
}
|
||||
}
|
||||
|
||||
return this.runFastInstanceCommand({ command, name });
|
||||
return this.runFastInstanceCommand({
|
||||
command,
|
||||
name,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
+126
-64
@@ -10,6 +10,12 @@ import {
|
||||
} from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
|
||||
import { formatUpgradeErrorForStorage } from 'src/engine/core-modules/upgrade/utils/format-upgrade-error-for-storage.util';
|
||||
|
||||
export type WorkspaceCursor = {
|
||||
name: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
isInitial: boolean;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class UpgradeMigrationService {
|
||||
constructor(
|
||||
@@ -35,73 +41,96 @@ export class UpgradeMigrationService {
|
||||
return isDefined(latestAttempt) && latestAttempt.status === 'completed';
|
||||
}
|
||||
|
||||
async markAsCompleted({
|
||||
name,
|
||||
workspaceId,
|
||||
executedByVersion,
|
||||
queryRunner,
|
||||
}: {
|
||||
name: string;
|
||||
workspaceId: string | null;
|
||||
executedByVersion: string;
|
||||
queryRunner?: QueryRunner;
|
||||
}): Promise<void> {
|
||||
const repository = queryRunner
|
||||
? queryRunner.manager.getRepository(UpgradeMigrationEntity)
|
||||
async recordUpgradeMigration(
|
||||
params:
|
||||
| {
|
||||
name: string;
|
||||
workspaceIds: string[];
|
||||
isInstance: boolean;
|
||||
status: 'completed';
|
||||
executedByVersion: string;
|
||||
queryRunner?: QueryRunner;
|
||||
}
|
||||
| {
|
||||
name: string;
|
||||
workspaceIds: string[];
|
||||
isInstance: boolean;
|
||||
status: 'failed';
|
||||
executedByVersion: string;
|
||||
error: unknown;
|
||||
queryRunner?: QueryRunner;
|
||||
},
|
||||
): Promise<void> {
|
||||
const { name, workspaceIds, isInstance, status, executedByVersion } =
|
||||
params;
|
||||
|
||||
const repository = params.queryRunner
|
||||
? params.queryRunner.manager.getRepository(UpgradeMigrationEntity)
|
||||
: this.upgradeMigrationRepository;
|
||||
const previousAttempts = await repository.count({
|
||||
where: {
|
||||
name,
|
||||
workspaceId: workspaceId === null ? IsNull() : workspaceId,
|
||||
},
|
||||
});
|
||||
|
||||
await repository.save({
|
||||
name,
|
||||
status: 'completed',
|
||||
attempt: previousAttempts + 1,
|
||||
executedByVersion,
|
||||
workspaceId,
|
||||
});
|
||||
const errorMessage =
|
||||
params.status === 'failed'
|
||||
? formatUpgradeErrorForStorage(params.error)
|
||||
: null;
|
||||
|
||||
if (isInstance) {
|
||||
const previousAttempts = await repository.count({
|
||||
where: { name, workspaceId: IsNull() },
|
||||
});
|
||||
|
||||
await repository.save([
|
||||
{
|
||||
name,
|
||||
status,
|
||||
attempt: previousAttempts + 1,
|
||||
executedByVersion,
|
||||
workspaceId: null,
|
||||
errorMessage,
|
||||
},
|
||||
...workspaceIds.map((workspaceId) => ({
|
||||
name,
|
||||
status,
|
||||
attempt: previousAttempts + 1,
|
||||
executedByVersion,
|
||||
workspaceId,
|
||||
errorMessage,
|
||||
})),
|
||||
]);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const rows = [];
|
||||
|
||||
for (const workspaceId of workspaceIds) {
|
||||
const previousAttempts = await repository.count({
|
||||
where: { name, workspaceId },
|
||||
});
|
||||
|
||||
rows.push({
|
||||
name,
|
||||
status,
|
||||
attempt: previousAttempts + 1,
|
||||
executedByVersion,
|
||||
workspaceId,
|
||||
errorMessage,
|
||||
});
|
||||
}
|
||||
|
||||
await repository.save(rows);
|
||||
}
|
||||
|
||||
async markAsFailed({
|
||||
name,
|
||||
workspaceId,
|
||||
executedByVersion,
|
||||
error,
|
||||
}: {
|
||||
name: string;
|
||||
workspaceId: string | null;
|
||||
executedByVersion: string;
|
||||
error: unknown;
|
||||
}): Promise<void> {
|
||||
const previousAttempts = await this.upgradeMigrationRepository.count({
|
||||
where: {
|
||||
name,
|
||||
workspaceId: workspaceId === null ? IsNull() : workspaceId,
|
||||
},
|
||||
});
|
||||
|
||||
await this.upgradeMigrationRepository.save({
|
||||
name,
|
||||
status: 'failed',
|
||||
attempt: previousAttempts + 1,
|
||||
executedByVersion,
|
||||
workspaceId,
|
||||
errorMessage: formatUpgradeErrorForStorage(error),
|
||||
});
|
||||
}
|
||||
|
||||
async markAsInitial({
|
||||
async markAsWorkspaceInitial({
|
||||
name,
|
||||
workspaceId,
|
||||
executedByVersion,
|
||||
status,
|
||||
queryRunner,
|
||||
}: {
|
||||
name: string;
|
||||
workspaceId: string;
|
||||
executedByVersion: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
queryRunner?: QueryRunner;
|
||||
}): Promise<void> {
|
||||
const repository = queryRunner
|
||||
@@ -110,7 +139,7 @@ export class UpgradeMigrationService {
|
||||
|
||||
await repository.save({
|
||||
name,
|
||||
status: 'completed',
|
||||
status,
|
||||
isInitial: true,
|
||||
attempt: 1,
|
||||
executedByVersion,
|
||||
@@ -120,8 +149,8 @@ export class UpgradeMigrationService {
|
||||
|
||||
// Returns the most recently attempted command (by createdAt)
|
||||
// across instance and active-workspace scopes, with its status.
|
||||
// Workspace-scoped records from inactive/deleted workspaces are
|
||||
// excluded so they cannot incorrectly influence the global cursor.
|
||||
// isInitial records are excluded — they represent activation
|
||||
// state, not execution progress.
|
||||
async getLastAttemptedCommandNameOrThrow(
|
||||
allActiveOrSuspendedWorkspaceIds: string[],
|
||||
): Promise<{
|
||||
@@ -131,6 +160,7 @@ export class UpgradeMigrationService {
|
||||
const queryBuilder = this.upgradeMigrationRepository
|
||||
.createQueryBuilder('migration')
|
||||
.select(['migration.name', 'migration.status'])
|
||||
.andWhere('migration."isInitial" = false')
|
||||
.andWhere(
|
||||
`migration.attempt = (
|
||||
SELECT MAX(sub.attempt)
|
||||
@@ -167,7 +197,7 @@ export class UpgradeMigrationService {
|
||||
|
||||
async getWorkspaceLastAttemptedCommandNameOrThrow(
|
||||
workspaceIds: string[],
|
||||
): Promise<Map<string, { name: string; status: UpgradeMigrationStatus }>> {
|
||||
): Promise<Map<string, WorkspaceCursor>> {
|
||||
if (workspaceIds.length === 0) {
|
||||
return new Map();
|
||||
}
|
||||
@@ -177,6 +207,7 @@ export class UpgradeMigrationService {
|
||||
.select('migration.workspaceId', 'workspaceId')
|
||||
.addSelect('migration.name', 'name')
|
||||
.addSelect('migration.status', 'status')
|
||||
.addSelect('migration.isInitial', 'isInitial')
|
||||
.where({
|
||||
workspaceId: In(workspaceIds),
|
||||
})
|
||||
@@ -195,15 +226,17 @@ export class UpgradeMigrationService {
|
||||
workspaceId: string;
|
||||
name: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
isInitial: boolean;
|
||||
}>();
|
||||
|
||||
const cursors = new Map<
|
||||
string,
|
||||
{ name: string; status: UpgradeMigrationStatus }
|
||||
>();
|
||||
const cursors = new Map<string, WorkspaceCursor>();
|
||||
|
||||
for (const row of results) {
|
||||
cursors.set(row.workspaceId, { name: row.name, status: row.status });
|
||||
cursors.set(row.workspaceId, {
|
||||
name: row.name,
|
||||
status: row.status,
|
||||
isInitial: row.isInitial,
|
||||
});
|
||||
}
|
||||
|
||||
const missingWorkspaceIds = workspaceIds.filter(
|
||||
@@ -249,4 +282,33 @@ export class UpgradeMigrationService {
|
||||
|
||||
return completedCount === workspaceIds.length;
|
||||
}
|
||||
|
||||
async getLastAttemptedInstanceCommandOrThrow(): Promise<{
|
||||
name: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
}> {
|
||||
const migration = await this.upgradeMigrationRepository
|
||||
.createQueryBuilder('migration')
|
||||
.select(['migration.name', 'migration.status'])
|
||||
.where('migration."workspaceId" IS NULL')
|
||||
.andWhere('migration."isInitial" = false')
|
||||
.andWhere(
|
||||
`migration.attempt = (
|
||||
SELECT MAX(sub.attempt)
|
||||
FROM core."upgradeMigration" sub
|
||||
WHERE sub.name = migration.name
|
||||
AND sub."workspaceId" IS NULL
|
||||
)`,
|
||||
)
|
||||
.orderBy('migration.createdAt', 'DESC')
|
||||
.getOne();
|
||||
|
||||
if (!migration) {
|
||||
throw new Error(
|
||||
'No instance command found — the database may not have been initialized',
|
||||
);
|
||||
}
|
||||
|
||||
return { name: migration.name, status: migration.status };
|
||||
}
|
||||
}
|
||||
|
||||
+39
-10
@@ -7,6 +7,8 @@ import {
|
||||
type RegisteredWorkspaceCommand,
|
||||
UpgradeCommandRegistryService,
|
||||
} from 'src/engine/core-modules/upgrade/services/upgrade-command-registry.service';
|
||||
import { type UpgradeMigrationStatus } from 'src/engine/core-modules/upgrade/upgrade-migration.entity';
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
export type FastInstanceUpgradeStep = {
|
||||
kind: 'fast-instance';
|
||||
@@ -78,7 +80,7 @@ export class UpgradeSequenceReaderService {
|
||||
return cursor;
|
||||
}
|
||||
|
||||
getWorkspaceCommandsSliceBounds({
|
||||
getWorkspaceSegmentBounds({
|
||||
sequence,
|
||||
workspaceCommand,
|
||||
}: {
|
||||
@@ -108,7 +110,7 @@ export class UpgradeSequenceReaderService {
|
||||
return { startCursor, endCursor };
|
||||
}
|
||||
|
||||
collectContiguousWorkspaceSteps({
|
||||
collectWorkspaceCommandsStartingFrom({
|
||||
sequence,
|
||||
fromWorkspaceCommand,
|
||||
}: {
|
||||
@@ -160,19 +162,46 @@ export class UpgradeSequenceReaderService {
|
||||
: workspaceCommands.slice(cursorIndex);
|
||||
}
|
||||
|
||||
getLastWorkspaceCommand(): RegisteredWorkspaceCommand {
|
||||
getInitialCursorForNewWorkspace(lastAttemptedInstanceCommand: {
|
||||
name: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
}): {
|
||||
name: string;
|
||||
status: UpgradeMigrationStatus;
|
||||
} {
|
||||
const { name, status } = lastAttemptedInstanceCommand;
|
||||
const sequence = this.getUpgradeSequence();
|
||||
|
||||
for (let index = sequence.length - 1; index >= 0; index--) {
|
||||
const step = sequence[index];
|
||||
const instanceCursor = this.locateStepInSequenceOrThrow({
|
||||
sequence,
|
||||
stepName: name,
|
||||
});
|
||||
|
||||
if (step.kind === 'workspace') {
|
||||
return step;
|
||||
if (status === 'completed') {
|
||||
const nextStep = sequence[instanceCursor + 1];
|
||||
|
||||
if (isDefined(nextStep) && nextStep.kind === 'workspace') {
|
||||
const lastWc = this.findLastWorkspaceCommandInSegmentStartingAt(
|
||||
sequence,
|
||||
nextStep,
|
||||
);
|
||||
|
||||
return { name: lastWc.name, status: 'completed' };
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(
|
||||
'No workspace commands found in upgrade sequence — this should have been caught at startup',
|
||||
);
|
||||
return { name, status };
|
||||
}
|
||||
|
||||
private findLastWorkspaceCommandInSegmentStartingAt(
|
||||
sequence: UpgradeStep[],
|
||||
firstWorkspaceCommand: WorkspaceUpgradeStep,
|
||||
): RegisteredWorkspaceCommand {
|
||||
const segment = this.collectWorkspaceCommandsStartingFrom({
|
||||
sequence,
|
||||
fromWorkspaceCommand: firstWorkspaceCommand,
|
||||
});
|
||||
|
||||
return segment[segment.length - 1];
|
||||
}
|
||||
}
|
||||
|
||||
+109
-41
@@ -6,7 +6,10 @@ import {
|
||||
} from 'src/database/commands/command-runners/workspace-iterator.service';
|
||||
import { type ParsedUpgradeCommandOptions } from 'src/database/commands/upgrade-version-command/upgrade.command';
|
||||
import { InstanceCommandRunnerService } from 'src/engine/core-modules/upgrade/services/instance-command-runner.service';
|
||||
import { UpgradeMigrationService } from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
|
||||
import {
|
||||
type WorkspaceCursor,
|
||||
UpgradeMigrationService,
|
||||
} from 'src/engine/core-modules/upgrade/services/upgrade-migration.service';
|
||||
import {
|
||||
type InstanceUpgradeStep,
|
||||
type UpgradeStep,
|
||||
@@ -57,16 +60,21 @@ export class UpgradeSequenceRunnerService {
|
||||
let totalSuccesses = 0;
|
||||
let totalFailures = 0;
|
||||
let cursor = startCursor;
|
||||
let workspaceCursors = await this.fetchWorkspaceCursors(
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
);
|
||||
|
||||
while (cursor < sequence.length) {
|
||||
const step = sequence[cursor];
|
||||
|
||||
if (step.kind === 'fast-instance' || step.kind === 'slow-instance') {
|
||||
const previousStep = cursor > 0 ? sequence[cursor - 1] : undefined;
|
||||
|
||||
if (previousStep?.kind === 'workspace') {
|
||||
await this.enforceWorkspaceSyncBarrier({
|
||||
this.enforceWorkspacesCompletedPreviousWorkspaceSegment({
|
||||
sequence,
|
||||
previousWorkspaceStep: previousStep,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
workspaceCursors,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -74,18 +82,20 @@ export class UpgradeSequenceRunnerService {
|
||||
instanceStep: step,
|
||||
skipDataMigration: allActiveOrSuspendedWorkspaceIds.length === 0,
|
||||
});
|
||||
|
||||
cursor++;
|
||||
continue;
|
||||
}
|
||||
|
||||
const contiguousWorkspaceSteps =
|
||||
this.upgradeSequenceReaderService.collectContiguousWorkspaceSteps({
|
||||
const workspaceCommandsSegment =
|
||||
this.upgradeSequenceReaderService.collectWorkspaceCommandsStartingFrom({
|
||||
sequence,
|
||||
fromWorkspaceCommand: step,
|
||||
});
|
||||
|
||||
const report = await this.resumeWorkspaceCommandsFromCursors({
|
||||
contiguousWorkspaceSteps,
|
||||
workspaceCommandsSegment,
|
||||
workspaceCursors,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
options,
|
||||
});
|
||||
@@ -102,11 +112,16 @@ export class UpgradeSequenceRunnerService {
|
||||
return { totalSuccesses, totalFailures };
|
||||
}
|
||||
|
||||
cursor += contiguousWorkspaceSteps.length;
|
||||
cursor += workspaceCommandsSegment.length;
|
||||
|
||||
workspaceCursors = await this.fetchWorkspaceCursors(
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
);
|
||||
}
|
||||
|
||||
return { totalSuccesses, totalFailures };
|
||||
}
|
||||
|
||||
private async resolveStartCursor({
|
||||
sequence,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
@@ -136,12 +151,12 @@ export class UpgradeSequenceRunnerService {
|
||||
}
|
||||
case 'workspace': {
|
||||
const workspaceSliceBounds =
|
||||
this.upgradeSequenceReaderService.getWorkspaceCommandsSliceBounds({
|
||||
this.upgradeSequenceReaderService.getWorkspaceSegmentBounds({
|
||||
sequence,
|
||||
workspaceCommand: lastAttemptedStep,
|
||||
});
|
||||
|
||||
await this.validateWorkspaceCursorsAreInSameWorkspaceStepsSlice({
|
||||
await this.validateWorkspaceCursorsAreInWorkspaceSegment({
|
||||
sequence,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
workspaceSliceBounds,
|
||||
@@ -154,7 +169,7 @@ export class UpgradeSequenceRunnerService {
|
||||
}
|
||||
}
|
||||
|
||||
private async validateWorkspaceCursorsAreInSameWorkspaceStepsSlice({
|
||||
private async validateWorkspaceCursorsAreInWorkspaceSegment({
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
sequence,
|
||||
workspaceSliceBounds: { startCursor, endCursor },
|
||||
@@ -167,22 +182,61 @@ export class UpgradeSequenceRunnerService {
|
||||
await this.upgradeMigrationService.getWorkspaceLastAttemptedCommandNameOrThrow(
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
);
|
||||
const precedingStep =
|
||||
startCursor > 0 ? sequence[startCursor - 1] : undefined;
|
||||
|
||||
const invalidWorkspaces: Array<{
|
||||
workspaceId: string;
|
||||
cursorName: string;
|
||||
cursorStatus: string;
|
||||
}> = [];
|
||||
|
||||
for (const [workspaceId, workspaceCursor] of workspaceCursors) {
|
||||
const cursor =
|
||||
const cursorPosition =
|
||||
this.upgradeSequenceReaderService.locateStepInSequenceOrThrow({
|
||||
sequence,
|
||||
stepName: workspaceCursor.name,
|
||||
});
|
||||
|
||||
if (cursor < startCursor || cursor > endCursor) {
|
||||
throw new Error(
|
||||
`Workspace ${workspaceId} cursor "${workspaceCursor.name}" is outside the ` +
|
||||
`current workspace slice [${startCursor}..${endCursor}] — ` +
|
||||
'workspaces are not aligned',
|
||||
);
|
||||
const isWithinSegment =
|
||||
cursorPosition >= startCursor && cursorPosition <= endCursor;
|
||||
|
||||
const isAtPrecedingInstanceCommandCompleted =
|
||||
isDefined(precedingStep) &&
|
||||
precedingStep.kind !== 'workspace' &&
|
||||
cursorPosition === startCursor - 1 &&
|
||||
workspaceCursor.status === 'completed';
|
||||
|
||||
if (!isWithinSegment && !isAtPrecedingInstanceCommandCompleted) {
|
||||
invalidWorkspaces.push({
|
||||
workspaceId,
|
||||
cursorName: workspaceCursor.name,
|
||||
cursorStatus: workspaceCursor.status,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (invalidWorkspaces.length > 0) {
|
||||
const details = invalidWorkspaces
|
||||
.map(
|
||||
({ workspaceId, cursorName, cursorStatus }) =>
|
||||
`${workspaceId} at "${cursorName}" (${cursorStatus})`,
|
||||
)
|
||||
.join(', ');
|
||||
|
||||
throw new Error(
|
||||
`${invalidWorkspaces.length} workspace(s) have invalid cursors for ` +
|
||||
`workspace segment [${startCursor}..${endCursor}]: ${details}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchWorkspaceCursors(
|
||||
allActiveOrSuspendedWorkspaceIds: string[],
|
||||
): Promise<Map<string, WorkspaceCursor>> {
|
||||
return this.upgradeMigrationService.getWorkspaceLastAttemptedCommandNameOrThrow(
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
);
|
||||
}
|
||||
|
||||
private async runInstanceStep({
|
||||
@@ -226,24 +280,23 @@ export class UpgradeSequenceRunnerService {
|
||||
}
|
||||
|
||||
private async resumeWorkspaceCommandsFromCursors({
|
||||
contiguousWorkspaceSteps,
|
||||
workspaceCommandsSegment,
|
||||
workspaceCursors,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
options,
|
||||
}: {
|
||||
contiguousWorkspaceSteps: WorkspaceUpgradeStep[];
|
||||
workspaceCommandsSegment: WorkspaceUpgradeStep[];
|
||||
workspaceCursors: Map<string, WorkspaceCursor>;
|
||||
allActiveOrSuspendedWorkspaceIds: string[];
|
||||
options: ParsedUpgradeCommandOptions;
|
||||
}): Promise<WorkspaceIteratorReport> {
|
||||
const workspaceCursors =
|
||||
await this.upgradeMigrationService.getWorkspaceLastAttemptedCommandNameOrThrow(
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
);
|
||||
const workspaceIds =
|
||||
isDefined(options.workspaceIds) && options.workspaceIds.length > 0
|
||||
? options.workspaceIds
|
||||
: allActiveOrSuspendedWorkspaceIds;
|
||||
|
||||
return this.workspaceIteratorService.iterate({
|
||||
workspaceIds:
|
||||
isDefined(options.workspaceIds) && options.workspaceIds.length > 0
|
||||
? options.workspaceIds
|
||||
: allActiveOrSuspendedWorkspaceIds,
|
||||
workspaceIds,
|
||||
startFromWorkspaceId: options.startFromWorkspaceId,
|
||||
workspaceCountLimit: options.workspaceCountLimit,
|
||||
dryRun: options.dryRun,
|
||||
@@ -258,7 +311,7 @@ export class UpgradeSequenceRunnerService {
|
||||
|
||||
const pendingCommands =
|
||||
this.upgradeSequenceReaderService.getPendingWorkspaceCommands({
|
||||
workspaceCommands: contiguousWorkspaceSteps,
|
||||
workspaceCommands: workspaceCommandsSegment,
|
||||
workspaceCursor,
|
||||
});
|
||||
|
||||
@@ -271,24 +324,39 @@ export class UpgradeSequenceRunnerService {
|
||||
});
|
||||
}
|
||||
|
||||
private async enforceWorkspaceSyncBarrier({
|
||||
private enforceWorkspacesCompletedPreviousWorkspaceSegment({
|
||||
sequence,
|
||||
previousWorkspaceStep,
|
||||
allActiveOrSuspendedWorkspaceIds,
|
||||
workspaceCursors,
|
||||
}: {
|
||||
sequence: UpgradeStep[];
|
||||
previousWorkspaceStep: WorkspaceUpgradeStep;
|
||||
allActiveOrSuspendedWorkspaceIds: string[];
|
||||
}): Promise<void> {
|
||||
const allWorkspacesReady =
|
||||
await this.upgradeMigrationService.areAllWorkspacesAtCommand({
|
||||
commandName: previousWorkspaceStep.name,
|
||||
workspaceIds: allActiveOrSuspendedWorkspaceIds,
|
||||
workspaceCursors: Map<string, WorkspaceCursor>;
|
||||
}): void {
|
||||
const barrierCursor =
|
||||
this.upgradeSequenceReaderService.locateStepInSequenceOrThrow({
|
||||
sequence,
|
||||
stepName: previousWorkspaceStep.name,
|
||||
});
|
||||
|
||||
if (!allWorkspacesReady) {
|
||||
throw new Error(
|
||||
'Cannot run instance step: not all workspaces have completed ' +
|
||||
`"${previousWorkspaceStep.name}"`,
|
||||
);
|
||||
for (const [workspaceId, workspaceCursor] of workspaceCursors) {
|
||||
const cursorPosition =
|
||||
this.upgradeSequenceReaderService.locateStepInSequenceOrThrow({
|
||||
sequence,
|
||||
stepName: workspaceCursor.name,
|
||||
});
|
||||
|
||||
const isAtBarrierAndCompleted =
|
||||
cursorPosition === barrierCursor &&
|
||||
workspaceCursor.status === 'completed';
|
||||
|
||||
if (!isAtBarrierAndCompleted) {
|
||||
throw new Error(
|
||||
`Cannot run instance step: workspace ${workspaceId} ` +
|
||||
`has not completed "${previousWorkspaceStep.name}" ` +
|
||||
`(cursor: "${workspaceCursor.name}", status: "${workspaceCursor.status}")`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+8
-4
@@ -78,17 +78,21 @@ export class WorkspaceCommandRunnerService {
|
||||
});
|
||||
|
||||
if (!options.dryRun) {
|
||||
await this.upgradeMigrationService.markAsCompleted({
|
||||
await this.upgradeMigrationService.recordUpgradeMigration({
|
||||
name,
|
||||
workspaceId,
|
||||
workspaceIds: [workspaceId],
|
||||
isInstance: false,
|
||||
status: 'completed',
|
||||
executedByVersion,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
if (!options.dryRun) {
|
||||
await this.upgradeMigrationService.markAsFailed({
|
||||
await this.upgradeMigrationService.recordUpgradeMigration({
|
||||
name,
|
||||
workspaceId,
|
||||
workspaceIds: [workspaceId],
|
||||
isInstance: false,
|
||||
status: 'failed',
|
||||
executedByVersion,
|
||||
error,
|
||||
});
|
||||
|
||||
+10
-4
@@ -383,8 +383,13 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
|
||||
workspaceId: string;
|
||||
displayName: string;
|
||||
}): Promise<void> {
|
||||
const lastWorkspaceCommand =
|
||||
this.upgradeSequenceReaderService.getLastWorkspaceCommand();
|
||||
const lastAttemptedInstanceCommand =
|
||||
await this.upgradeMigrationService.getLastAttemptedInstanceCommandOrThrow();
|
||||
|
||||
const initialCursor =
|
||||
this.upgradeSequenceReaderService.getInitialCursorForNewWorkspace(
|
||||
lastAttemptedInstanceCommand,
|
||||
);
|
||||
|
||||
const executedByVersion =
|
||||
this.twentyConfigService.get('APP_VERSION') ?? 'unknown';
|
||||
@@ -400,10 +405,11 @@ export class WorkspaceService extends TypeOrmQueryService<WorkspaceEntity> {
|
||||
activationStatus: WorkspaceActivationStatus.ACTIVE,
|
||||
});
|
||||
|
||||
await this.upgradeMigrationService.markAsInitial({
|
||||
name: lastWorkspaceCommand.name,
|
||||
await this.upgradeMigrationService.markAsWorkspaceInitial({
|
||||
name: initialCursor.name,
|
||||
workspaceId,
|
||||
executedByVersion,
|
||||
status: initialCursor.status,
|
||||
queryRunner,
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user