[pull] main from activepieces:main

.devcontainer/Dockerfile

@@ -31,7 +31,7 @@ ENV LC_ALL en_US.UTF-8
 
 COPY default.cf /usr/local/etc/isolate
 
-RUN npm i -g bun@1.3.1 cross-env@7.0.3 esbuild@0.25.0 @anthropic-ai/claude-code
+RUN npm i -g bun@1.4.0 cross-env@7.0.3 esbuild@0.25.0 @anthropic-ai/claude-code
 
 
 # install isolated-vm in a parent directory to avoid linking the package in every sandbox

Dockerfile

@@ -34,9 +34,9 @@ RUN apt-get update && \
 # Download, extract, and clean up bun in a single layer so the zip never ships
 RUN export ARCH=$(uname -m) && \
     if [ "$ARCH" = "x86_64" ]; then \
-      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-x64-baseline.zip -o bun.zip; \
+      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-x64-baseline.zip -o bun.zip; \
     elif [ "$ARCH" = "aarch64" ]; then \
-      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-aarch64.zip -o bun.zip; \
+      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-aarch64.zip -o bun.zip; \
     fi && \
     unzip bun.zip && \
     mv bun-*/bun /usr/local/bin/bun && \

Dockerfile.worker

@@ -13,9 +13,9 @@ RUN apt-get update && \
 
 RUN export ARCH=$(uname -m) && \
     if [ "$ARCH" = "x86_64" ]; then \
-      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-x64-baseline.zip -o bun.zip; \
+      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-x64-baseline.zip -o bun.zip; \
     elif [ "$ARCH" = "aarch64" ]; then \
-      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-aarch64.zip -o bun.zip; \
+      curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-aarch64.zip -o bun.zip; \
     fi && \
     unzip bun.zip && mv bun-*/bun /usr/local/bin/bun && chmod +x /usr/local/bin/bun && rm -rf bun.zip bun-*
 

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   app:
-    image: ghcr.io/activepieces/activepieces:0.88.3
+    image: ghcr.io/activepieces/activepieces:0.89.0
     container_name: activepieces-app
     restart: unless-stopped
     ports:
@@ -16,7 +16,7 @@ services:
     networks:
       - activepieces
   worker:
-    image: ghcr.io/activepieces/activepieces:0.88.3
+    image: ghcr.io/activepieces/activepieces:0.89.0
     restart: unless-stopped
     depends_on:
       - app

package.json

@@ -1,7 +1,7 @@
 {
   "name": "activepieces",
-  "version": "0.88.3",
-  "packageManager": "bun@1.3.3",
+  "version": "0.89.0",
+  "packageManager": "bun@1.4.0",
   "trustedDependencies": [
     "sqlite3",
     "isolated-vm",

packages/core/execution/src/lib/flows/dto/create-flow-request.ts

@@ -1,5 +1,5 @@
 import { z } from 'zod'
-import { Metadata } from '@activepieces/core-utils'
+import { formErrors, Metadata, SAFE_EXTERNAL_ID_PATTERN } from '@activepieces/core-utils'
 
 export const CreateFlowRequest = z.object({
     displayName: z.string(),
@@ -9,6 +9,7 @@ export const CreateFlowRequest = z.object({
     projectId: z.string(),
     templateId: z.string().optional(),
     metadata: z.optional(Metadata),
+    externalId: z.string().regex(SAFE_EXTERNAL_ID_PATTERN, formErrors.invalidExternalId).optional(),
 })
 
 export type CreateFlowRequest = z.infer<typeof CreateFlowRequest>

packages/core/shared/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@activepieces/shared",
-  "version": "0.151.0",
+  "version": "0.152.0",
   "type": "commonjs",
   "sideEffects": false,
   "main": "./dist/src/index.js",

packages/core/shared/src/lib/automation/tables/dto/tables.dto.ts

@@ -1,12 +1,10 @@
-import { Nullable, OptionalArrayFromQuery } from '@activepieces/core-utils'
+import { Nullable, OptionalArrayFromQuery, SAFE_EXTERNAL_ID_PATTERN } from '@activepieces/core-utils'
 import { z } from 'zod'
 import { formErrors } from '../../../form-errors'
 import { FieldState } from '../../project-release/project-state'
 import { TableAutomationStatus, TableAutomationTrigger } from '../table'
 import { TableWebhookEventType } from '../table-webhook'
 
-const SAFE_EXTERNAL_ID_PATTERN = /^(?!\.{1,2}$)[A-Za-z0-9._-]{1,128}$/
-
 export const CreateTableRequest = z.object({
     projectId: z.string(),
     name: z.string(),

packages/core/utils/src/lib/activepieces-error.ts

@@ -68,6 +68,7 @@ export type ApErrorParams =
     | InvalidGitCredentialsParams
     | InvalidReleaseTypeParams
     | ProjectExternalIdAlreadyExistsParams
+    | FlowExternalIdAlreadyExistsParams
     | SandboxMemoryIssueParams
     | SandboxExecutionTimeoutParams
     | SandboxInternalErrorParams
@@ -426,6 +427,10 @@ export type ProjectExternalIdAlreadyExistsParams = BaseErrorParams<ErrorCode.PRO
     externalId: string
 }>
 
+export type FlowExternalIdAlreadyExistsParams = BaseErrorParams<ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS, {
+    externalId: string
+}>
+
 export type McpPieceRequiresConnectionParams = BaseErrorParams<ErrorCode.MCP_PIECE_REQUIRES_CONNECTION, {
     pieceName: string
 }>
@@ -523,6 +528,7 @@ export enum ErrorCode {
     EXISTING_USER = 'EXISTING_USER',
     EXISTING_ALERT_CHANNEL = 'EXISTING_ALERT_CHANNEL',
     PROJECT_EXTERNAL_ID_ALREADY_EXISTS = 'PROJECT_EXTERNAL_ID_ALREADY_EXISTS',
+    FLOW_EXTERNAL_ID_ALREADY_EXISTS = 'FLOW_EXTERNAL_ID_ALREADY_EXISTS',
     FLOW_OPERATION_INVALID = 'FLOW_OPERATION_INVALID',
     FLOW_OPERATION_IN_PROGRESS = 'FLOW_OPERATION_IN_PROGRESS',
     FLOW_RUN_RETRY_OUTSIDE_RETENTION = 'FLOW_RUN_RETRY_OUTSIDE_RETENTION',

packages/core/utils/src/lib/form-errors.ts

@@ -7,3 +7,5 @@ export const formErrors = {
     invalidFileName: 'invalidFileName',
     messageRequiresContentOrFiles: 'messageRequiresContentOrFiles',
 } as const
+
+export const SAFE_EXTERNAL_ID_PATTERN = /^(?!\.{1,2}$)[A-Za-z0-9._-]{1,128}$/

packages/server/api/src/app/database/migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex.ts

@@ -0,0 +1,44 @@
+import { QueryRunner } from 'typeorm'
+import { system } from '../../../helper/system/system'
+import { AppSystemProp } from '../../../helper/system/system-props'
+import { DatabaseType } from '../../database-type'
+import { Migration } from '../../migration'
+
+export class AddFlowProjectIdExternalIdUniqueIndex1839000000000 implements Migration {
+    name = 'AddFlowProjectIdExternalIdUniqueIndex1839000000000'
+    breaking = false
+    release = '0.88.4'
+    transaction = false
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        const concurrently = isPGlite() ? '' : 'CONCURRENTLY'
+
+        await queryRunner.query(`
+            UPDATE "flow"
+            SET "externalId" = "id"
+            WHERE "id" IN (
+                SELECT "id" FROM (
+                    SELECT "id",
+                           ROW_NUMBER() OVER (
+                               PARTITION BY "projectId", "externalId"
+                               ORDER BY "updated" DESC, "created" DESC, "id" DESC
+                           ) AS rn
+                    FROM "flow"
+                ) ranked
+                WHERE ranked.rn > 1
+            )
+        `)
+
+        await queryRunner.query(`
+            CREATE UNIQUE INDEX ${concurrently} IF NOT EXISTS "idx_flow_project_id_external_id"
+            ON "flow" ("projectId", "externalId")
+        `)
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        const concurrently = isPGlite() ? '' : 'CONCURRENTLY'
+        await queryRunner.query(`DROP INDEX ${concurrently} IF EXISTS "idx_flow_project_id_external_id"`)
+    }
+}
+
+const isPGlite = (): boolean => system.get(AppSystemProp.DB_TYPE) === DatabaseType.PGLITE

packages/server/api/src/app/database/postgres-connection.ts

@@ -431,6 +431,7 @@ import { WidenMcpOAuthState1835000000000 } from './migration/postgres/1835000000
 import { DropTeamsBotInstallation1836000000000 } from './migration/postgres/1836000000000-DropTeamsBotInstallation'
 import { AddAiProviderStatus1837000000000 } from './migration/postgres/1837000000000-AddAiProviderStatus'
 import { AddMcpOAuthTokenLastUsedAndClientKey1838000000000 } from './migration/postgres/1838000000000-AddMcpOAuthTokenLastUsedAndClientKey'
+import { AddFlowProjectIdExternalIdUniqueIndex1839000000000 } from './migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex'
 
 const getSslConfig = (): boolean | TlsOptions => {
     const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL)
@@ -877,6 +878,7 @@ export const getMigrations = (): (new () => Migration)[] => {
         DropTeamsBotInstallation1836000000000,
         AddAiProviderStatus1837000000000,
         AddMcpOAuthTokenLastUsedAndClientKey1838000000000,
+        AddFlowProjectIdExternalIdUniqueIndex1839000000000,
     ]
     return migrations
 }

packages/server/api/src/app/file/file.service.ts

@@ -69,6 +69,9 @@ export const fileService = (log: FastifyBaseLogger) => ({
                     return await fileRepo().save({ ...baseFile, location: FileLocation.S3, s3Key })
                 }
                 catch (error) {
+                    if (isNil(params.data)) {
+                        throw error
+                    }
                     exceptionHandler.handle(error, log)
                     return saveFileToDb(baseFile, params.data)
                 }

packages/server/api/src/app/flows/flow/flow.controller.ts

@@ -28,6 +28,7 @@ export const flowController: FastifyPluginAsyncZod = async (app) => {
             ownerId: actorUserId(request),
             templateId: request.body.templateId,
             ip: networkUtils.clientIp(request),
+            externalId: request.body.externalId,
         })
 
         return reply.status(StatusCodes.CREATED).send(newFlow)

packages/server/api/src/app/flows/flow/flow.entity.ts

@@ -101,6 +101,11 @@ export const FlowEntity = new EntitySchema<FlowSchema>({
             columns: ['projectId', 'status'],
             unique: false,
         },
+        {
+            name: 'idx_flow_project_id_external_id',
+            columns: ['projectId', 'externalId'],
+            unique: true,
+        },
     ],
     relations: {
         runs: {

packages/server/api/src/app/flows/flow/flow.service.ts

@@ -30,6 +30,7 @@ import { flowRepo } from './flow.repo'
 
 export const flowService = (log: FastifyBaseLogger) => ({
     async create({ projectId, request, externalId, ownerId, templateId, createdBy, ip, emitEvents = true }: CreateParams): Promise<PopulatedFlow> {
+        await assertExternalIdIsUnique({ projectId, externalId })
         const folderId = await getFolderIdFromRequest({ projectId, folderId: request.folderId, folderName: request.folderName, log })
         const newFlow: NewFlow = {
             id: apId(),
@@ -802,6 +803,19 @@ const assertFlowIsNotNull: <T extends Flow>(
     }
 }
 
+async function assertExternalIdIsUnique({ projectId, externalId }: { projectId: ProjectId, externalId: string | undefined }): Promise<void> {
+    if (isNil(externalId)) {
+        return
+    }
+    const exists = await flowRepo().existsBy({ projectId, externalId })
+    if (exists) {
+        throw new ActivepiecesError({
+            code: ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS,
+            params: { externalId },
+        })
+    }
+}
+
 type CreateParams = EventEmissionParams & {
     projectId: ProjectId
     request: CreateFlowRequest

packages/server/api/src/app/helper/error-handler.ts

@@ -91,6 +91,7 @@ const statusCodeMap: Partial<Record<ErrorCode, StatusCodes>> = {
     [ErrorCode.AUTHORIZATION]: StatusCodes.FORBIDDEN,
     [ErrorCode.SIGN_UP_DISABLED]: StatusCodes.FORBIDDEN,
     [ErrorCode.PROJECT_EXTERNAL_ID_ALREADY_EXISTS]: StatusCodes.CONFLICT,
+    [ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS]: StatusCodes.CONFLICT,
     [ErrorCode.INVALID_CREDENTIALS]: StatusCodes.UNAUTHORIZED,
     [ErrorCode.SESSION_EXPIRED]: StatusCodes.FORBIDDEN,
     [ErrorCode.EMAIL_IS_NOT_VERIFIED]: StatusCodes.FORBIDDEN,

packages/server/api/src/app/workers/rpc/worker-rpc-service.ts

@@ -3,7 +3,7 @@ import { apVersionUtil, onCallService, UNKNOWN_VERSION } from '@activepieces/ser
 import { ExecutionType, FileCompression, FileLocation, FileType, FlowOperationType, FlowStatus, WebsocketClientEvent, WorkerGroupScope, WorkerToApiContract } from '@activepieces/shared'
 import { FastifyBaseLogger } from 'fastify'
 import { websocketService } from '../../core/websockets.service'
-import { redisConnections } from '../../database/redis-connections'
+import { distributedStore, redisConnections } from '../../database/redis-connections'
 import { agentRpcHandlers } from '../../ee/agent/agent-rpc-handlers'
 import { chatPersonalizationService } from '../../ee/agent/personalization/chat-personalization-service'
 import { fileService, getLocationForFile } from '../../file/file.service'
@@ -29,6 +29,10 @@ import { getPlatformGroupQueueName, getProjectGroupQueueName, QueueName, WorkerG
 import { jobBroker } from '../job-queue/job-broker'
 import { machineService } from '../machine/machine-service'
 
+const FLOW_BUNDLE_PUBLISH_CLAIM_TTL_SECONDS = 60
+
+const getFlowBundlePublishClaimKey = (flowVersionId: string): string => `flow_bundle_publish_claim:${flowVersionId}`
+
 const getPollQueueName = (assignment: WorkerGroupAssignment | null): string => {
     if (isNil(assignment)) {
         return QueueName.WORKER_JOBS
@@ -237,6 +241,16 @@ export function createHandlers(log: FastifyBaseLogger, assignment: WorkerGroupAs
             if (getLocationForFile(FileType.FLOW_BUNDLE) !== FileLocation.S3) {
                 return { kind: 'skip' }
             }
+            // The bundle for a flowVersionId is immutable, and a burst of workers can
+            // finish provisioning the same version at once. Only the first claimer
+            // publishes; losing the claim is a normal outcome, not an error. Without
+            // this, concurrent callers race the deterministic file PK (duplicate key
+            // on the file table) and hammer the same S3 key (503 Slow Down). The
+            // claim expires so a publish that died mid-upload gets retried.
+            const claimed = await distributedStore.putIfAbsent(getFlowBundlePublishClaimKey(input.flowVersionId), 1, FLOW_BUNDLE_PUBLISH_CLAIM_TTL_SECONDS)
+            if (!claimed) {
+                return { kind: 'skip' }
+            }
             // S3 without signed URLs: the worker streams the bytes back via uploadFlowBundle.
             if (!signedFileTransport.shouldRedirectForType(FileType.FLOW_BUNDLE)) {
                 return { kind: 'inline' }

packages/web/public/locales/en/translation.json

@@ -2304,6 +2304,7 @@
   "Highlights": "Highlights",
   "External ID can only contain letters, numbers and underscores": "External ID can only contain letters, numbers and underscores",
   "required": "Required",
+  "invalidExternalId": "External ID may only contain letters, numbers, dot, dash and underscore (up to 128 characters)",
   "Please connect your account first": "Please connect your account first",
   "Recommended": "Recommended",
   "Others": "Others",