[One Workflow] Implement cancel-in-progress collision strategy (elastic#247217)

Author: h88

src/platform/packages/shared/kbn-workflows/spec/lib/generate_yaml_schema_from_connectors.ts

@@ -41,15 +41,13 @@ export function generateYamlSchemaFromConnectors(
 
   if (loose) {
     return WorkflowSchema.partial().extend({
-      settings: WorkflowSettingsSchema.omit({ concurrency: true }).optional(),
+      settings: WorkflowSettingsSchema.optional(),
       steps: z.array(recursiveStepSchema).optional(),
     });
   }
 
   return WorkflowSchema.extend({
-    settings: getWorkflowSettingsSchema(recursiveStepSchema, loose)
-      .omit({ concurrency: true })
-      .optional(),
+    settings: getWorkflowSettingsSchema(recursiveStepSchema, loose).optional(),
     steps: z.array(recursiveStepSchema),
   });
 }

src/platform/packages/shared/kbn-workflows/spec/schema.test.ts

@@ -180,7 +180,8 @@ describe('ConcurrencySettingsSchema', () => {
 
   describe('strategy', () => {
     it('should accept valid strategy values', () => {
-      const strategies = ['queue', 'drop', 'cancel-in-progress'] as const;
+      // Only 'cancel-in-progress' is currently implemented; 'queue' and 'drop' are TBD
+      const strategies = ['cancel-in-progress'] as const;
       strategies.forEach((strategy) => {
         const result = ConcurrencySettingsSchema.safeParse({
           strategy,
@@ -263,7 +264,7 @@ describe('ConcurrencySettingsSchema', () => {
     // Verify the type can be used and matches the schema inference
     const testSettings: ConcurrencySettings = {
       key: '{{ event.host.name }}',
-      strategy: 'queue',
+      strategy: 'cancel-in-progress',
       max: 3,
     };
     const result = ConcurrencySettingsSchema.safeParse(testSettings);
@@ -282,14 +283,14 @@ describe('WorkflowSettingsSchema', () => {
       const result = WorkflowSettingsSchema.safeParse({
         concurrency: {
           key: '{{ event.host.name }}',
-          strategy: 'queue',
+          strategy: 'cancel-in-progress',
           max: 3,
         },
       });
       expect(result.success).toBe(true);
       if (result.success) {
         expect(result.data.concurrency?.key).toBe('{{ event.host.name }}');
-        expect(result.data.concurrency?.strategy).toBe('queue');
+        expect(result.data.concurrency?.strategy).toBe('cancel-in-progress');
         expect(result.data.concurrency?.max).toBe(3);
       }
     });
@@ -328,9 +329,10 @@ describe('WorkflowSettingsSchema', () => {
 
   describe('CollisionStrategySchema', () => {
     it('should accept all valid strategy values', () => {
-      expect(CollisionStrategySchema.safeParse('queue').success).toBe(true);
-      expect(CollisionStrategySchema.safeParse('drop').success).toBe(true);
+      // Only 'cancel-in-progress' is currently implemented; 'queue' and 'drop' are TBD
       expect(CollisionStrategySchema.safeParse('cancel-in-progress').success).toBe(true);
+      expect(CollisionStrategySchema.safeParse('queue').success).toBe(false);
+      expect(CollisionStrategySchema.safeParse('drop').success).toBe(false);
     });
 
     it('should reject invalid strategy values', () => {
@@ -341,7 +343,7 @@ describe('WorkflowSettingsSchema', () => {
 
     it('should export CollisionStrategy type that matches valid values', () => {
       // Verify the type can be used and matches the schema values
-      const validStrategies: CollisionStrategy[] = ['queue', 'drop', 'cancel-in-progress'];
+      const validStrategies: CollisionStrategy[] = ['cancel-in-progress']; // 'queue', 'drop',  TBD
       validStrategies.forEach((strategy) => {
         const result = CollisionStrategySchema.safeParse(strategy);
         expect(result.success).toBe(true);

src/platform/packages/shared/kbn-workflows/spec/schema.ts

@@ -56,7 +56,7 @@ export function getOnFailureStepSchema(stepSchema: z.ZodType, loose: boolean = f
   return schema;
 }
 
-export const CollisionStrategySchema = z.enum(['queue', 'drop', 'cancel-in-progress']);
+export const CollisionStrategySchema = z.enum(['cancel-in-progress']); // 'queue', 'drop' TBD
 export type CollisionStrategy = z.infer<typeof CollisionStrategySchema>;
 
 export const ConcurrencySettingsSchema = z.object({

src/platform/packages/shared/kbn-workflows/types/latest.ts

@@ -63,6 +63,7 @@ export {
   ExecutionType,
   ExecutionStatusValues,
   ExecutionTypeValues,
+  TerminalExecutionStatuses,
   SearchWorkflowCommandSchema,
   UpdateWorkflowCommandSchema,
 } from './v1';

src/platform/packages/shared/kbn-workflows/types/utils.ts

@@ -14,7 +14,7 @@ import type {
   HttpMethod,
   InternalConnectorContract,
 } from './v1';
-import { ExecutionStatus, KNOWN_HTTP_METHODS } from './v1';
+import { ExecutionStatus, KNOWN_HTTP_METHODS, TerminalExecutionStatuses } from './v1';
 import type {
   BuiltInStepType,
   ElasticsearchStep,
@@ -60,14 +60,7 @@ export function isDangerousStatus(status: ExecutionStatus) {
 }
 
 export function isTerminalStatus(status: ExecutionStatus) {
-  const TerminalStatus: readonly ExecutionStatus[] = [
-    ExecutionStatus.COMPLETED,
-    ExecutionStatus.FAILED,
-    ExecutionStatus.CANCELLED,
-    ExecutionStatus.SKIPPED,
-    ExecutionStatus.TIMED_OUT,
-  ];
-  return TerminalStatus.includes(status);
+  return TerminalExecutionStatuses.includes(status);
 }
 
 export function isCancelableStatus(status: ExecutionStatus) {

src/platform/packages/shared/kbn-workflows/types/v1.ts

@@ -29,6 +29,14 @@ export enum ExecutionStatus {
 export type ExecutionStatusUnion = `${ExecutionStatus}`;
 export const ExecutionStatusValues = Object.values(ExecutionStatus);
 
+export const TerminalExecutionStatuses: readonly ExecutionStatus[] = [
+  ExecutionStatus.COMPLETED,
+  ExecutionStatus.FAILED,
+  ExecutionStatus.CANCELLED,
+  ExecutionStatus.SKIPPED,
+  ExecutionStatus.TIMED_OUT,
+] as const;
+
 export enum ExecutionType {
   TEST = 'test',
   PRODUCTION = 'production',

src/platform/plugins/shared/workflows_execution_engine/integration_tests/mocks/workflow_execution_repository.mock.ts

@@ -8,7 +8,7 @@
  */
 
 import type { EsWorkflowExecution } from '@kbn/workflows';
-import { ExecutionStatus } from '@kbn/workflows';
+import { TerminalExecutionStatuses } from '@kbn/workflows';
 import type { WorkflowExecutionRepository as WorkflowExecutionRepositoryType } from '../../server/repositories/workflow_execution_repository';
 
 export class WorkflowExecutionRepositoryMock implements Required<WorkflowExecutionRepositoryType> {
@@ -102,19 +102,11 @@ export class WorkflowExecutionRepositoryMock implements Required<WorkflowExecuti
     spaceId: string,
     triggeredBy?: string
   ): Promise<Array<{ _source: EsWorkflowExecution; _id: string; _index: string }>> {
-    const terminalStatuses = [
-      ExecutionStatus.COMPLETED,
-      ExecutionStatus.FAILED,
-      ExecutionStatus.CANCELLED,
-      ExecutionStatus.SKIPPED,
-      ExecutionStatus.TIMED_OUT,
-    ];
-
     let results = Array.from(this.workflowExecutions.values()).filter(
       (exec) =>
         exec.workflowId === workflowId &&
         exec.spaceId === spaceId &&
-        !terminalStatuses.includes(exec.status)
+        !TerminalExecutionStatuses.includes(exec.status)
     );
 
     if (triggeredBy) {
@@ -128,4 +120,73 @@ export class WorkflowExecutionRepositoryMock implements Required<WorkflowExecuti
       _index: 'workflows-executions',
     }));
   }
+
+  public async getRunningExecutionsByConcurrencyGroup(
+    concurrencyGroupKey: string,
+    spaceId: string,
+    excludeExecutionId?: string,
+    size: number = 5000
+  ): Promise<string[]> {
+    const results = Array.from(this.workflowExecutions.values())
+      .filter(
+        (exec) =>
+          exec.concurrencyGroupKey === concurrencyGroupKey &&
+          exec.spaceId === spaceId &&
+          !TerminalExecutionStatuses.includes(exec.status) &&
+          (!excludeExecutionId || exec.id !== excludeExecutionId)
+      )
+      .sort((a, b) => {
+        const aTime = new Date(a.createdAt).getTime();
+        const bTime = new Date(b.createdAt).getTime();
+        return aTime - bTime; // Oldest first
+      })
+      .map((exec) => exec.id)
+      .slice(0, Math.min(size, 10000)); // Cap at ES default max_result_window
+
+    return results;
+  }
+
+  public async bulkUpdateWorkflowExecutions(
+    updates: Array<Partial<EsWorkflowExecution>>
+  ): Promise<void> {
+    if (updates.length === 0) {
+      return;
+    }
+
+    // Validate all IDs are present
+    for (const update of updates) {
+      if (!update.id) {
+        throw new Error('Workflow execution ID is required for bulk update');
+      }
+    }
+
+    // Validate all executions exist (matching Elasticsearch document_missing_exception behavior)
+    const missingIds: string[] = [];
+    for (const update of updates) {
+      if (!this.workflowExecutions.has(update.id!)) {
+        missingIds.push(update.id!);
+      }
+    }
+
+    if (missingIds.length > 0) {
+      throw new Error(
+        `Failed to update ${missingIds.length} workflow executions: ${JSON.stringify(
+          missingIds.map((id) => ({
+            id,
+            error: { type: 'document_missing_exception', reason: 'document missing' },
+            status: 404,
+          }))
+        )}`
+      );
+    }
+
+    // Perform updates
+    for (const update of updates) {
+      const existing = this.workflowExecutions.get(update.id!);
+      this.workflowExecutions.set(update.id!, {
+        ...existing!,
+        ...update,
+      } as EsWorkflowExecution);
+    }
+  }
 }