feat(workflow): Pick Record round robin strategy (2/3) (#21900)

## Overview Second PR in the Pick Record stack. Adds a **Round Robin** selection strategy alongside Random, so an assignment workflow can distribute records *evenly* across a candidate pool (e.g. rotate company ownership across a set of workspace members) rather than just randomly. **Stacked on #21899** — review/merge that one first. This PR's diff against `main` includes PR 1's commits until #21899 merges. ## What changed - Widened the `strategy` enum (`RANDOM` → `RANDOM | ROUND_ROBIN`) in the shared schema and the server input type. - Editor now shows a **Strategy** selector (Random / Round robin). The candidate-pool label changed from "Pick at random from" to the neutral "Pick from" since random is no longer the only mode. - Executor implements round robin. ## Design decisions & tradeoffs 1. **State store: Redis `incrBy` (atomic), keyed `pick-record:round-robin:{workspaceId}:{stepId}`.** Round robin needs a persistent cursor, and workflow runs are **not** serialized — two runs can execute the same step concurrently — so the increment must be atomic. `CacheStorageService.incrBy` (workflow cache namespace) is a single atomic Redis op, needs no schema change, and is already injectable. Index = `(cursor - 1) % poolSize`. **Tradeoff — durability:** a Redis flush/eviction resets the cursor, which restarts the cycle from an offset. That causes a one-time *fairness drift*, never a *correctness* bug (no double-assignment, since each increment is atomic). If strict durability is ever required, the cursor can move to a Postgres counter table with `INSERT … ON CONFLICT … DO UPDATE SET cursor = cursor + 1 RETURNING cursor` (atomic + durable) — deliberately **not** done here to avoid a migration for what is, in practice, an acceptable reset. 2. **Deterministic pool ordering.** The resolved pool is sorted by `id` before the cursor is applied, so position→record mapping is stable run-to-run regardless of fetch order. Without this, round robin wouldn't reliably cycle. 3. **Cursor key uses `stepId`.** Stable across runs of a published version. Republishing a version may mint new step ids, which resets the cursor — acceptable and documented here. 4. **Slot-on-increment.** The cursor increments when the step runs (reserving a position); if a later step in the run fails, that position is effectively skipped. Minor, acceptable unfairness — flagged rather than adding cross-step compensation. ## Testing Added `pick-record-round-robin-workflow.integration-spec.ts`: builds a workflow with a 3-record pool and `ROUND_ROBIN`, runs it 4 times sequentially, and asserts the picks are exactly `[p0, p1, p2, p0]` (full cycle + wraparound) against the deterministically-ordered pool. Passes locally alongside PR 1's random test (2 suites / 3 tests). `typecheck` + `lint:diff-with-main` green for shared/server/front. ## Follow-up - PR 3: `LOAD_BALANCED` (fewest related records wins). https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8 --- _Generated by [Claude Code](https://claude.ai/code/session_01MuPWZsqf2bmQSevk6fRbX8)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/21900?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-06-21 22:16:26 +02:00
parent a682c8fa62
commit fa6d1394af
5 changed files with 336 additions and 9 deletions
@@ -4,12 +4,14 @@ import { msg } from '@lingui/core/macro';
 import { useLingui } from '@lingui/react/macro';
 import { useEffect, useState } from 'react';
 import { isDefined } from 'twenty-shared/utils';
+import { type SelectOption } from 'twenty-ui/input';
 import { HorizontalSeparator } from 'twenty-ui/layout';
 import { useDebouncedCallback } from 'use-debounce';

 import { useFilteredObjectMetadataItems } from '@/object-metadata/hooks/useFilteredObjectMetadataItems';
 import { useObjectMetadataSelectHelpers } from '@/object-metadata/hooks/useObjectMetadataSelectHelpers';
 import { FormMultiRecordPicker } from '@/object-record/record-field/ui/form-types/components/FormMultiRecordPicker';
+import { Select } from '@/ui/input/components/Select';
 import { SelectControl } from '@/ui/input/components/SelectControl';
 import { Dropdown } from '@/ui/layout/dropdown/components/Dropdown';
 import { useCloseDropdown } from '@/ui/layout/dropdown/hooks/useCloseDropdown';
@@ -45,8 +47,12 @@ type WorkflowEditActionPickRecordProps = {
      };
 };

+type PickRecordStrategy =
+  WorkflowPickRecordAction['settings']['input']['strategy'];
+
 type PickRecordFormData = {
  objectNameSingular: string;
+  strategy: PickRecordStrategy;
  recordIds: string[];
 };

@@ -66,11 +72,17 @@ export const WorkflowEditActionPickRecord = ({

  const [formData, setFormData] = useState<PickRecordFormData>(() => ({
    objectNameSingular: action.settings.input.objectName,
+    strategy: action.settings.input.strategy,
    recordIds: action.settings.input.recordIds,
  }));

  const isFormDisabled = actionOptions.readonly ?? false;

+  const strategyOptions: SelectOption<PickRecordStrategy>[] = [
+    { label: t`Random`, value: 'RANDOM' },
+    { label: t`Round robin`, value: 'ROUND_ROBIN' },
+  ];
+
  const selectedObjectMetadataItem = objectMetadataItems.find(
    (item) => item.nameSingular === formData.objectNameSingular,
  );
@@ -94,7 +106,7 @@ export const WorkflowEditActionPickRecord = ({
          ...action.settings,
          input: {
            objectName: updatedFormData.objectNameSingular,
-            strategy: action.settings.input.strategy,
+            strategy: updatedFormData.strategy,
            recordIds: updatedFormData.recordIds,
          },
        },
@@ -115,6 +127,7 @@ export const WorkflowEditActionPickRecord = ({
    }

    const newFormData: PickRecordFormData = {
+      ...formData,
      objectNameSingular: value,
      recordIds: [],
    };
@@ -124,6 +137,20 @@ export const WorkflowEditActionPickRecord = ({
    closeDropdown(dropdownId);
  };

+  const handleStrategyChange = (strategy: PickRecordStrategy) => {
+    if (isFormDisabled === true) {
+      return;
+    }
+
+    const newFormData: PickRecordFormData = {
+      ...formData,
+      strategy,
+    };
+
+    setFormData(newFormData);
+    saveAction(newFormData);
+  };
+
  const handleRecordIdsChange = (recordIds: string[]) => {
    if (isFormDisabled === true) {
      return;
@@ -163,12 +190,22 @@ export const WorkflowEditActionPickRecord = ({
          />
        </StyledObjectSelectContainer>

+        <Select
+          dropdownId={`workflow-edit-action-pick-record-strategy-${action.id}`}
+          label={t`Strategy`}
+          fullWidth
+          disabled={isFormDisabled}
+          value={formData.strategy}
+          options={strategyOptions}
+          onChange={handleStrategyChange}
+        />
+
        <HorizontalSeparator noMargin />

        {isDefined(selectedObjectMetadataItem) && (
          <FormMultiRecordPicker
            key={selectedObjectMetadataItem.nameSingular}
-            label={t`Pick at random from`}
+            label={t`Pick from`}
            objectNameSingular={selectedObjectMetadataItem.nameSingular}
            defaultValue={formData.recordIds}
            onChange={(value) =>
@@ -6,6 +6,9 @@ import { resolveInput } from 'twenty-shared/utils';

 import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';

+import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
 import { FindRecordsService } from 'src/engine/core-modules/record-crud/services/find-records.service';
 import {
  WorkflowStepExecutorException,
@@ -16,13 +19,20 @@ import { type WorkflowActionInput } from 'src/modules/workflow/workflow-executor
 import { type WorkflowActionOutput } from 'src/modules/workflow/workflow-executor/types/workflow-action-output.type';
 import { findStepOrThrow } from 'src/modules/workflow/workflow-executor/utils/find-step-or-throw.util';
 import { isWorkflowPickRecordAction } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/guards/is-workflow-pick-record-action.guard';
-import { type WorkflowPickRecordActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type';
+import {
+  type WorkflowPickRecordActionInput,
+  type WorkflowPickRecordStrategy,
+} from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-pick-record-action-input.type';
+
+const ROUND_ROBIN_CURSOR_TTL_MS = 1000 * 60 * 60 * 24 * 90;

@Injectable()
 export class PickRecordWorkflowAction implements WorkflowAction {
  constructor(
    private readonly findRecordsService: FindRecordsService,
    private readonly workflowExecutionContextService: WorkflowExecutionContextService,
+    @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
+    private readonly cacheStorageService: CacheStorageService,
  ) {}

  async execute({
@@ -43,7 +53,7 @@ export class PickRecordWorkflowAction implements WorkflowAction {
      );
    }

-    const { objectName, recordIds } = resolveInput(
+    const { objectName, strategy, recordIds } = resolveInput(
      step.settings.input,
      context,
    ) as WorkflowPickRecordActionInput;
@@ -80,9 +90,43 @@ export class PickRecordWorkflowAction implements WorkflowAction {
      };
    }

-    const pickedRecord =
-      candidateRecords[Math.floor(Math.random() * candidateRecords.length)];
+    const orderedRecords = [...candidateRecords].sort((recordA, recordB) =>
+      String(recordA.id).localeCompare(String(recordB.id)),
+    );

-    return { result: pickedRecord };
+    const pickedIndex = await this.getPickedIndex({
+      strategy,
+      candidateCount: orderedRecords.length,
+      workspaceId: runInfo.workspaceId,
+      stepId: currentStepId,
+    });
+
+    return { result: orderedRecords[pickedIndex] };
+  }
+
+  private async getPickedIndex({
+    strategy,
+    candidateCount,
+    workspaceId,
+    stepId,
+  }: {
+    strategy: WorkflowPickRecordStrategy;
+    candidateCount: number;
+    workspaceId: string;
+    stepId: string;
+  }): Promise<number> {
+    if (strategy === 'ROUND_ROBIN') {
+      const cursorKey = `pick-record:round-robin:${workspaceId}:${stepId}`;
+      const nextCursor = await this.cacheStorageService.incrBy(cursorKey, 1);
+
+      await this.cacheStorageService.expire(
+        cursorKey,
+        ROUND_ROBIN_CURSOR_TTL_MS,
+      );
+
+      return (nextCursor - 1) % candidateCount;
+    }
+
+    return Math.floor(Math.random() * candidateCount);
  }
 }
@@ -1,4 +1,4 @@
-export type WorkflowPickRecordStrategy = 'RANDOM';
+export type WorkflowPickRecordStrategy = 'RANDOM' | 'ROUND_ROBIN';

 export type WorkflowPickRecordActionInput = {
  objectName: string;
@@ -0,0 +1,243 @@
+import request from 'supertest';
+import {
+  destroyWorkflowRun,
+  runWorkflowVersion,
+  waitForWorkflowCompletion,
+} from 'test/integration/graphql/suites/workflow/utils/workflow-run-test.util';
+
+const client = request(`http://localhost:${APP_PORT}`);
+
+describe('Pick Record Workflow - round robin (e2e)', () => {
+  let createdWorkflowId: string | null = null;
+  let createdWorkflowVersionId: string | null = null;
+  let pickRecordStepId: string | null = null;
+  let orderedCandidateRecordIds: string[] = [];
+
+  beforeAll(async () => {
+    const createWorkflowResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation CreateWorkflow {
+            createWorkflow(data: { name: "Pick Record Round Robin Test" }) {
+              id
+            }
+          }
+        `,
+      });
+
+    createdWorkflowId = createWorkflowResponse.body.data.createWorkflow.id;
+
+    const getWorkflowResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query GetWorkflow($id: UUID!) {
+            workflow(filter: { id: { eq: $id } }) {
+              versions {
+                edges {
+                  node {
+                    id
+                  }
+                }
+              }
+            }
+          }
+        `,
+        variables: { id: createdWorkflowId },
+      });
+
+    createdWorkflowVersionId =
+      getWorkflowResponse.body.data.workflow.versions.edges[0].node.id;
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation UpdateWorkflowVersion($id: UUID!, $data: WorkflowVersionUpdateInput!) {
+            updateWorkflowVersion(id: $id, data: $data) {
+              id
+            }
+          }
+        `,
+        variables: {
+          id: createdWorkflowVersionId,
+          data: {
+            trigger: {
+              name: 'Manual Trigger',
+              type: 'MANUAL',
+              settings: { outputSchema: {} },
+              nextStepIds: [],
+              position: { x: 0, y: 0 },
+            },
+          },
+        },
+      });
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation CreateWorkflowVersionStep($input: CreateWorkflowVersionStepInput!) {
+            createWorkflowVersionStep(input: $input) {
+              stepsDiff
+            }
+          }
+        `,
+        variables: {
+          input: {
+            workflowVersionId: createdWorkflowVersionId,
+            stepType: 'PICK_RECORD',
+            parentStepId: 'trigger',
+            position: { x: 200, y: 0 },
+          },
+        },
+      });
+
+    const getStepsResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query GetWorkflowVersion($id: UUID!) {
+            workflowVersion(filter: { id: { eq: $id } }) {
+              steps
+            }
+          }
+        `,
+        variables: { id: createdWorkflowVersionId },
+      });
+
+    const pickRecordStep =
+      getStepsResponse.body.data.workflowVersion.steps.find(
+        (step: { type: string }) => step.type === 'PICK_RECORD',
+      );
+
+    pickRecordStepId = pickRecordStep.id;
+
+    const companiesResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          query Companies {
+            companies(first: 3) {
+              edges {
+                node {
+                  id
+                }
+              }
+            }
+          }
+        `,
+      });
+
+    const candidateRecordIds = companiesResponse.body.data.companies.edges.map(
+      (edge: { node: { id: string } }) => edge.node.id,
+    );
+
+    expect(candidateRecordIds.length).toBe(3);
+
+    // The executor sorts the resolved pool deterministically by id before
+    // applying the round-robin cursor, so the expected cycle order is the
+    // pool sorted the same way.
+    orderedCandidateRecordIds = [...candidateRecordIds].sort(
+      (idA: string, idB: string) => idA.localeCompare(idB),
+    );
+
+    await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation UpdateWorkflowVersionStep($input: UpdateWorkflowVersionStepInput!) {
+            updateWorkflowVersionStep(input: $input) {
+              id
+            }
+          }
+        `,
+        variables: {
+          input: {
+            workflowVersionId: createdWorkflowVersionId,
+            step: {
+              ...pickRecordStep,
+              settings: {
+                ...pickRecordStep.settings,
+                input: {
+                  objectName: 'company',
+                  strategy: 'ROUND_ROBIN',
+                  recordIds: candidateRecordIds,
+                },
+              },
+            },
+          },
+        },
+      });
+
+    const activateResponse = await client
+      .post('/graphql')
+      .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+      .send({
+        query: `
+          mutation ActivateWorkflowVersion($workflowVersionId: UUID!) {
+            activateWorkflowVersion(workflowVersionId: $workflowVersionId)
+          }
+        `,
+        variables: { workflowVersionId: createdWorkflowVersionId },
+      });
+
+    expect(activateResponse.body.data.activateWorkflowVersion).toBe(true);
+  });
+
+  afterAll(async () => {
+    if (createdWorkflowId) {
+      await client
+        .post('/graphql')
+        .set('Authorization', `Bearer ${APPLE_JANE_ADMIN_ACCESS_TOKEN}`)
+        .send({
+          query: `
+            mutation DestroyWorkflow($id: ID!) {
+              destroyWorkflow(id: $id) {
+                id
+              }
+            }
+          `,
+          variables: { id: createdWorkflowId },
+        });
+    }
+  });
+
+  it('cycles through the pool in order on consecutive runs', async () => {
+    const pickedRecordIds: string[] = [];
+
+    for (let runIndex = 0; runIndex < 4; runIndex++) {
+      const workflowRunId = await runWorkflowVersion({
+        workflowVersionId: createdWorkflowVersionId!,
+        payload: {},
+      });
+
+      const workflowRun = await waitForWorkflowCompletion(workflowRunId);
+
+      expect(workflowRun?.status).toBe('COMPLETED');
+
+      const result = workflowRun?.state?.stepInfos?.[pickRecordStepId!]
+        ?.result as { id: string } | undefined;
+
+      expect(result?.id).toBeDefined();
+      pickedRecordIds.push(result!.id);
+
+      await destroyWorkflowRun(workflowRunId);
+    }
+
+    expect(pickedRecordIds).toEqual([
+      orderedCandidateRecordIds[0],
+      orderedCandidateRecordIds[1],
+      orderedCandidateRecordIds[2],
+      orderedCandidateRecordIds[0],
+    ]);
+  });
+});
@@ -1,7 +1,10 @@
 import { z } from 'zod';
 import { baseWorkflowActionSettingsSchema } from './base-workflow-action-settings-schema';

-export const workflowPickRecordStrategySchema = z.enum(['RANDOM']);
+export const workflowPickRecordStrategySchema = z.enum([
+  'RANDOM',
+  'ROUND_ROBIN',
+]);

 export const workflowPickRecordActionSettingsSchema =
  baseWorkflowActionSettingsSchema.extend({