Skip to content

Commit 13be0a3

Browse files
martmullmartmull
authored
feat(server): add command converting existing application logic functions to prebuilt (#25178)
Follow-up to #25119, which made *new* logic functions of *newly installed* packaged applications run in PREBUILT mode. That PR deliberately left already-installed applications on LIVE. This one backfills them. ## What it does Adds `upgrade:2-39:convert-logic-functions-to-prebuilt`, a registered workspace upgrade command. For every provisioned workspace it enables `IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED`, then converts that workspace's eligible logic functions from LIVE to PREBUILT. ``` upgrade:2-39:convert-logic-functions-to-prebuilt └─ per workspace enable IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED └─ collect eligible logic functions across every application └─ per batch of 100, Promise.allSettled └─ per logic function own transaction + bundle install ``` A logic function is eligible when: - its application comes from a packaged source (tarball or npm) and is not deleted - it is currently LIVE and not deleted - `isBuildUpToDate` is true and it carries a non-empty checksum Those last two mirror `isLogicFunctionReadyForPrebuiltInstall`, the invariant `FlatLogicFunctionValidatorService` enforces on PREBUILT rows. ## Conversion granularity The command calls `UpdateLogicFunctionActionHandlerService.executeForMetadata` directly rather than going through `validateBuildAndRunWorkspaceMigration`. That is what makes the parallelism possible: the migration path serialises one build-and-run per application, so the lambda installs ran application by application. Calling the handler directly lets `installPrebuiltBundleIfNeeded` run concurrently across logic functions. Consequences of bypassing the migration path, stated plainly: - No validator pass. The eligibility predicate mirrors the same invariant the validator enforces, and the handler re-checks through `shouldReinstallLogicFunctionPrebuiltBundle`, so the invariant holds, but it now rests on the predicate. - No metadata events and no optimistic cache update. `flatLogicFunctionMaps` is invalidated explicitly once the workspace finishes. Metadata version is unaffected: it only increments for object and field metadata. - The update partial is exactly `{ executionMode: PREBUILT }` rather than a full entity diffed against the builder's own read, which removes the stale-row replacement cubic raised on the earlier shape. ## The bundle is installed by the conversion, not on first execution ``` convertLogicFunctionToPrebuilt └─ queryRunner.startTransaction() └─ UpdateLogicFunctionActionHandlerService.executeForMetadata ├─ logicFunctionRepository.update (LIVE -> PREBUILT) └─ installPrebuiltBundleIfNeeded -> driver.installPrebuiltBundle ├─ getBuiltCode (built JS from file storage) ├─ createZipFile ├─ UpdateFunctionCode <- stores it in the lambda ├─ waitFunctionUpdated └─ TagResource (checksum tag) └─ commit ``` Nothing is *built* here: eligibility requires `isBuildUpToDate` and a checksum, so the artifact already exists in file storage and the drivers only install it. A function with a stale build is skipped rather than rebuilt. The on-demand installer from #25119 stays as the safety net for a node that missed the conversion. ## Failure and concurrency behaviour - **One transaction per logic function.** The row update and its bundle install commit together, so a failed install rolls that row back to LIVE and the next run reconsiders it. Nothing else in the batch is affected. - **One failure does not abort the rest.** `Promise.allSettled` per batch; rejections are logged per function and the workspace continues. - **A conversion failure does not fail the upgrade.** The command reports a summary and returns. It is idempotent: eligibility is keyed on `executionMode = LIVE`, and `installPrebuiltBundle` re-checks the installed checksum inside its lock. - **Rollback** is instant and needs no data change: with the feature flag off, `resolveEffectiveExecutionMode` forces LIVE regardless of the stored mode. **Open question for review.** A transaction per logic function means one pooled connection per in-flight conversion, held for the whole install. `PG_POOL_MAX_CONNECTIONS` defaults to 10 and a lambda update can run to 60s, so a batch of 100 does not run 100 conversions at once — 10 acquire connections and the rest queue, with the upgrade holding every core connection meanwhile. Effective concurrency is pool-bound. Worth deciding whether to lower the batch size, raise the pool for the upgrade process, or accept it. ## Note on the feature flag The upgrade enables `IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED` on every provisioned workspace, so it is no longer a brake on the rollout. Gated on a flag that is off almost everywhere, an upgrade command would be a no-op for most workspaces. The flag is only written where it is not already set, since enabling invalidates and recomputes the workspace cache. ## The workflowVersion guard `cec8ed74` also fixes `WorkspaceWorkflowAutomatedTriggerMapCacheService.findWorkspaceVersionIdByCoreVersionId`, which is a **separate logical change** and is called out as such. It queried the workspace `workflowVersion` object on `coreWorkflowVersionId`, a field that does not exist on a database upgrading from before the workflow-core migration (#25104). The query threw, and one workspace's failed cache recompute aborted the entire upgrade sequence — `cross-version-upgrade` reported 70 workspaces succeeded, 2 failed, everything stopped. It is guarded the way `WorkflowVersionCoreSyncService.workspaceHasCoreWorkflowVersionIdField` already guards this exact field; the map degrades to a null workspace twin id, which `computeAutomatedTriggerFromWorkflowVersion` already accepts. It no-ops once an equivalent guard lands on main. It is bundled here because `cross-version-upgrade` could not go green on this branch without it. If #25104's authors would rather own it, lifting those 27 lines into their own PR and dropping the commit here is clean. ## Test plan - `is-logic-function-eligible-for-prebuilt-conversion.util.spec.ts`: 9 cases covering both packaged sources, both unpackaged sources, already-prebuilt, stale build, null and empty checksum, soft-deleted - 95 tests pass across the workflow and logic-function suites, including main's own `compute-automated-trigger-from-workflow-version` spec which exercises the degraded path - `nx build twenty-server`, `tsgo --noEmit`, `oxlint --type-aware` and `oxfmt --check` all clean - an earlier revision was tested locally end to end by @martmull Full CI is green on `cec8ed74`, including `cross-version-upgrade` and `server-validation`. Not yet run against production data, and the 100-way concurrent `UpdateFunctionCode` has not been exercised against real lambda control-plane limits. ## History This started as a standalone `logic-function:convert-to-prebuilt` command enqueuing per-application jobs on `logicFunctionQueue`. After @Weiko and @prastoin it became an upgrade command with inline conversion; after @martmull and @prastoin the conversion service was folded into the command, then re-cut from per-application migrations to per-logic-function transactions. The queue-related findings from the earlier shape (job priority, retry policy, `bulkAdd` bounds, `lockDuration`) no longer apply.
1 parent 28f6e37 commit 13be0a3

7 files changed

Lines changed: 414 additions & 0 deletions

File tree

‎packages/twenty-server/src/database/commands/upgrade-version-command/2-39/2-39-upgrade-version-command.module.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
import { Module } from '@nestjs/common';
2+
3+
import { WorkspaceIteratorModule } from 'src/database/commands/command-runners/workspace-iterator.module';
4+
import { ConvertLogicFunctionsToPrebuiltCommand } from 'src/database/commands/upgrade-version-command/2-39/2-39-workspace-command-1788338950836-convert-logic-functions-to-prebuilt.command';
5+
import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
6+
import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
7+
import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
8+
import { WorkspaceSchemaMigrationRunnerActionHandlersModule } from 'src/engine/workspace-manager/workspace-migration/workspace-migration-runner/action-handlers/workspace-schema-migration-runner-action-handlers.module';
9+
10+
@Module({
11+
imports: [
12+
FeatureFlagModule,
13+
TypeORMModule,
14+
WorkspaceIteratorModule,
15+
WorkspaceManyOrAllFlatEntityMapsCacheModule,
16+
WorkspaceSchemaMigrationRunnerActionHandlersModule,
17+
],
18+
providers: [ConvertLogicFunctionsToPrebuiltCommand],
19+
})
20+
export class V2_39_UpgradeVersionCommandModule {}

‎packages/twenty-server/src/database/commands/upgrade-version-command/2-39/2-39-workspace-command-1788338950836-convert-logic-functions-to-prebuilt.command.ts‎

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,251 @@
1+
import { InjectDataSource } from '@nestjs/typeorm';
2+
3+
import chunk from 'lodash.chunk';
4+
import { Command } from 'nest-commander';
5+
6+
import { FeatureFlagKey } from 'twenty-shared/types';
7+
import { isDefined } from 'twenty-shared/utils';
8+
import { DataSource } from 'typeorm';
9+
10+
import { ProvisionedWorkspaceCommandRunner } from 'src/database/commands/command-runners/provisioned-workspace.command-runner';
11+
import { WorkspaceIteratorService } from 'src/database/commands/command-runners/workspace-iterator.service';
12+
import { type RunOnWorkspaceArgs } from 'src/database/commands/command-runners/workspace.command-runner';
13+
import { type FlatApplicationCacheMaps } from 'src/engine/core-modules/application/types/flat-application-cache-maps.type';
14+
import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type';
15+
import { findActiveFlatApplicationById } from 'src/engine/core-modules/application/utils/find-active-flat-application-by-id.util';
16+
import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
17+
import { RegisteredWorkspaceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-workspace-command.decorator';
18+
import { createEmptyAllFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/constant/create-empty-all-flat-entity-maps.constant';
19+
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
20+
import { type AllFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/types/all-flat-entity-maps.type';
21+
import { LogicFunctionExecutionMode } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
22+
import { type FlatLogicFunctionMaps } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function-maps.type';
23+
import { type FlatLogicFunction } from 'src/engine/metadata-modules/logic-function/types/flat-logic-function.type';
24+
import { isLogicFunctionEligibleForPrebuiltConversion } from 'src/engine/metadata-modules/logic-function/utils/is-logic-function-eligible-for-prebuilt-conversion.util';
25+
import { UpdateLogicFunctionActionHandlerService } from 'src/engine/workspace-manager/workspace-migration/workspace-migration-runner/action-handlers/logic-function/services/update-logic-function-action-handler.service';
26+
27+
const LOGIC_FUNCTION_PREBUILT_CONVERSION_BATCH_SIZE = 100;
28+
29+
type LogicFunctionConversionTarget = {
30+
flatLogicFunction: FlatLogicFunction;
31+
flatApplication: FlatApplication;
32+
};
33+
34+
@RegisteredWorkspaceCommand('2.39.0', 1788338950836)
35+
@Command({
36+
name: 'upgrade:2-39:convert-logic-functions-to-prebuilt',
37+
description:
38+
'Convert packaged application logic functions from LIVE to PREBUILT execution mode. Idempotent.',
39+
})
40+
export class ConvertLogicFunctionsToPrebuiltCommand extends ProvisionedWorkspaceCommandRunner {
41+
constructor(
42+
protected readonly workspaceIteratorService: WorkspaceIteratorService,
43+
private readonly featureFlagService: FeatureFlagService,
44+
private readonly flatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
45+
private readonly updateLogicFunctionActionHandlerService: UpdateLogicFunctionActionHandlerService,
46+
@InjectDataSource()
47+
private readonly coreDataSource: DataSource,
48+
) {
49+
super(workspaceIteratorService);
50+
}
51+
52+
override async runOnWorkspace({
53+
workspaceId,
54+
options,
55+
}: RunOnWorkspaceArgs): Promise<void> {
56+
const dryRun = options.dryRun ?? false;
57+
58+
const isPrebuiltModeEnabled =
59+
await this.featureFlagService.isFeatureEnabled(
60+
FeatureFlagKey.IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED,
61+
workspaceId,
62+
);
63+
64+
if (!isPrebuiltModeEnabled && !dryRun) {
65+
await this.featureFlagService.enableFeatureFlags(
66+
[FeatureFlagKey.IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED],
67+
workspaceId,
68+
);
69+
}
70+
71+
const { flatLogicFunctionMaps, flatApplicationMaps } =
72+
await this.flatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps(
73+
{
74+
workspaceId,
75+
flatMapsKeys: ['flatLogicFunctionMaps', 'flatApplicationMaps'],
76+
},
77+
);
78+
79+
const conversionTargets = this.findLogicFunctionsToConvert({
80+
flatLogicFunctionMaps,
81+
flatApplicationMaps,
82+
});
83+
84+
if (conversionTargets.length === 0) {
85+
return;
86+
}
87+
88+
if (dryRun) {
89+
this.logger.log(
90+
`Would ensure ${FeatureFlagKey.IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED} is enabled and convert ${conversionTargets.length} logic function(s) on workspace ${workspaceId}`,
91+
);
92+
93+
return;
94+
}
95+
96+
const convertedCount = await this.convertLogicFunctionsInBatches({
97+
conversionTargets,
98+
workspaceId,
99+
allFlatEntityMaps: {
100+
...createEmptyAllFlatEntityMaps(),
101+
flatLogicFunctionMaps,
102+
},
103+
});
104+
105+
if (convertedCount > 0) {
106+
await this.flatEntityMapsCacheService.invalidateFlatEntityMaps({
107+
workspaceId,
108+
flatMapsKeys: ['flatLogicFunctionMaps'],
109+
});
110+
}
111+
}
112+
113+
private async convertLogicFunctionsInBatches({
114+
conversionTargets,
115+
workspaceId,
116+
allFlatEntityMaps,
117+
}: {
118+
conversionTargets: LogicFunctionConversionTarget[];
119+
workspaceId: string;
120+
allFlatEntityMaps: AllFlatEntityMaps;
121+
}): Promise<number> {
122+
let convertedCount = 0;
123+
const failedLogicFunctionIds: string[] = [];
124+
125+
for (const batch of chunk(
126+
conversionTargets,
127+
LOGIC_FUNCTION_PREBUILT_CONVERSION_BATCH_SIZE,
128+
)) {
129+
const results = await Promise.allSettled(
130+
batch.map((conversionTarget) =>
131+
this.convertLogicFunctionToPrebuilt({
132+
conversionTarget,
133+
workspaceId,
134+
allFlatEntityMaps,
135+
}),
136+
),
137+
);
138+
139+
results.forEach((result, batchIndex) => {
140+
if (result.status === 'fulfilled') {
141+
convertedCount += 1;
142+
143+
return;
144+
}
145+
146+
const { flatLogicFunction } = batch[batchIndex];
147+
148+
failedLogicFunctionIds.push(flatLogicFunction.id);
149+
150+
this.logger.error(
151+
`Failed to convert logic function '${flatLogicFunction.id}' on workspace ${workspaceId}: ${
152+
result.reason instanceof Error
153+
? result.reason.message
154+
: String(result.reason)
155+
}`,
156+
);
157+
});
158+
}
159+
160+
this.logger.log(
161+
`Converted ${convertedCount} logic function(s) on workspace ${workspaceId}` +
162+
(failedLogicFunctionIds.length > 0
163+
? `, ${failedLogicFunctionIds.length} failed: ${failedLogicFunctionIds.join(', ')}`
164+
: ''),
165+
);
166+
167+
return convertedCount;
168+
}
169+
170+
private async convertLogicFunctionToPrebuilt({
171+
conversionTarget: { flatLogicFunction, flatApplication },
172+
workspaceId,
173+
allFlatEntityMaps,
174+
}: {
175+
conversionTarget: LogicFunctionConversionTarget;
176+
workspaceId: string;
177+
allFlatEntityMaps: AllFlatEntityMaps;
178+
}): Promise<void> {
179+
const queryRunner = this.coreDataSource.createQueryRunner();
180+
181+
try {
182+
await queryRunner.connect();
183+
await queryRunner.startTransaction();
184+
185+
await this.updateLogicFunctionActionHandlerService.executeForMetadata({
186+
queryRunner,
187+
workspaceId,
188+
allFlatEntityMaps,
189+
flatApplication,
190+
action: {
191+
type: 'update',
192+
metadataName: 'logicFunction',
193+
universalIdentifier: flatLogicFunction.universalIdentifier,
194+
update: { executionMode: LogicFunctionExecutionMode.PREBUILT },
195+
},
196+
flatAction: {
197+
type: 'update',
198+
metadataName: 'logicFunction',
199+
entityId: flatLogicFunction.id,
200+
update: { executionMode: LogicFunctionExecutionMode.PREBUILT },
201+
},
202+
});
203+
204+
await queryRunner.commitTransaction();
205+
} catch (error) {
206+
if (queryRunner.isTransactionActive) {
207+
await queryRunner.rollbackTransaction();
208+
}
209+
210+
throw error;
211+
} finally {
212+
await queryRunner.release();
213+
}
214+
}
215+
216+
private findLogicFunctionsToConvert({
217+
flatLogicFunctionMaps,
218+
flatApplicationMaps,
219+
}: {
220+
flatLogicFunctionMaps: FlatLogicFunctionMaps;
221+
flatApplicationMaps: FlatApplicationCacheMaps;
222+
}): LogicFunctionConversionTarget[] {
223+
return Object.entries(
224+
flatLogicFunctionMaps.universalIdentifiersByApplicationId,
225+
).flatMap(([applicationId, universalIdentifiers]) => {
226+
const flatApplication = findActiveFlatApplicationById(
227+
flatApplicationMaps,
228+
applicationId,
229+
);
230+
231+
if (!isDefined(flatApplication)) {
232+
return [];
233+
}
234+
235+
return (universalIdentifiers ?? [])
236+
.map(
237+
(universalIdentifier) =>
238+
flatLogicFunctionMaps.byUniversalIdentifier[universalIdentifier],
239+
)
240+
.filter(
241+
(flatLogicFunction): flatLogicFunction is FlatLogicFunction =>
242+
isDefined(flatLogicFunction) &&
243+
isLogicFunctionEligibleForPrebuiltConversion({
244+
flatLogicFunction,
245+
applicationSourceType: flatApplication.sourceType,
246+
}),
247+
)
248+
.map((flatLogicFunction) => ({ flatLogicFunction, flatApplication }));
249+
});
250+
}
251+
}

‎packages/twenty-server/src/database/commands/upgrade-version-command/workspace-command-provider.module.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import { V2_35_UpgradeVersionCommandModule } from 'src/database/commands/upgrade
3131
import { V2_36_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-36/2-36-upgrade-version-command.module';
3232
import { V2_37_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-37/2-37-upgrade-version-command.module';
3333
import { V2_38_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-38/2-38-upgrade-version-command.module';
34+
import { V2_39_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-39/2-39-upgrade-version-command.module';
3435
import { V2_4_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-4/2-4-upgrade-version-command.module';
3536
import { V2_5_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-5/2-5-upgrade-version-command.module';
3637
import { V2_7_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/2-7/2-7-upgrade-version-command.module';
@@ -75,6 +76,7 @@ import { V2_9_UpgradeVersionCommandModule } from 'src/database/commands/upgrade-
7576
V2_36_UpgradeVersionCommandModule,
7677
V2_37_UpgradeVersionCommandModule,
7778
V2_38_UpgradeVersionCommandModule,
79+
V2_39_UpgradeVersionCommandModule,
7880
],
7981
})
8082
export class WorkspaceCommandProviderModule {}

‎packages/twenty-server/src/engine/core-modules/workflow/services/workspace-workflow-automated-trigger-map-cache.service.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { Injectable, Logger } from '@nestjs/common';
22

3+
import { STANDARD_OBJECTS } from 'twenty-shared/metadata';
34
import { isDefined } from 'twenty-shared/utils';
45
import { In } from 'typeorm';
56

@@ -14,6 +15,7 @@ import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system
1415
import { InjectWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/inject-workspace-scoped-repository.decorator';
1516
import { WorkspaceScopedRepository } from 'src/engine/twenty-orm/workspace-scoped-repository/workspace-scoped-repository';
1617
import { WorkspaceCache } from 'src/engine/workspace-cache/decorators/workspace-cache.decorator';
18+
import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
1719
import { WorkspaceCacheProvider } from 'src/engine/workspace-cache/interfaces/workspace-cache-provider.service';
1820
import { type WorkspaceCacheProviderContext } from 'src/engine/workspace-cache/types/workspace-cache-provider-context.type';
1921
import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
@@ -29,6 +31,7 @@ export class WorkspaceWorkflowAutomatedTriggerMapCacheService extends WorkspaceC
2931
@InjectWorkspaceScopedRepository(WorkflowVersionEntity)
3032
private readonly workflowVersionRepository: WorkspaceScopedRepository<WorkflowVersionEntity>,
3133
private readonly workspaceOrmManager: WorkspaceOrmManager,
34+
private readonly workspaceCacheService: WorkspaceCacheService,
3235
) {
3336
super();
3437
}
@@ -75,6 +78,14 @@ export class WorkspaceWorkflowAutomatedTriggerMapCacheService extends WorkspaceC
7578
return {};
7679
}
7780

81+
if (!(await this.workspaceHasCoreWorkflowVersionIdField(workspaceId))) {
82+
this.logger.warn(
83+
`workflowVersion.coreWorkflowVersionId field missing for workspace ${workspaceId}, skipping workspace twin resolution`,
84+
);
85+
86+
return {};
87+
}
88+
7889
const authContext = buildSystemAuthContext(workspaceId);
7990

8091
return this.workspaceOrmManager.executeInWorkspaceContext(async () => {
@@ -143,4 +154,20 @@ export class WorkspaceWorkflowAutomatedTriggerMapCacheService extends WorkspaceC
143154
);
144155
}, authContext);
145156
}
157+
158+
private async workspaceHasCoreWorkflowVersionIdField(
159+
workspaceId: string,
160+
): Promise<boolean> {
161+
const { flatFieldMetadataMaps } =
162+
await this.workspaceCacheService.getOrRecompute(workspaceId, [
163+
'flatFieldMetadataMaps',
164+
]);
165+
166+
return isDefined(
167+
flatFieldMetadataMaps.byUniversalIdentifier[
168+
STANDARD_OBJECTS.workflowVersion.fields.coreWorkflowVersionId
169+
.universalIdentifier
170+
],
171+
);
172+
}
146173
}

0 commit comments

Comments
 (0)