feat(core): atomizer with nx:atomizer executor -- WIP

packages/cypress/src/plugins/plugin.ts

@@ -295,7 +295,7 @@ async function buildCypressTargets(
       }
 
       targets[options.ciTargetName] = {
-        executor: 'nx:noop',
+        executor: 'nx:atomizer',
         cache: true,
         inputs,
         outputs,

packages/jest/src/plugins/plugin.ts

@@ -242,7 +242,7 @@ async function buildJestTargets(
       const dependsOn: string[] = [];
 
       targets[options.ciTargetName] = {
-        executor: 'nx:noop',
+        executor: 'nx:atomizer',
         cache: true,
         inputs,
         outputs,

packages/nx/executors.json

@@ -14,6 +14,11 @@
       "implementation": "./src/executors/run-script/run-script.impl",
       "schema": "./src/executors/run-script/schema.json",
       "description": "Run an NPM script using Nx."
+    },
+    "atomizer": {
+      "implementation": "./src/executors/atomizer/atomizer.impl",
+      "schema": "./src/executors/atomizer/schema.json",
+      "description": "An executor that is used as the root task for atomized tasks. Ensures it's being run in distribution and otherwise does nothing."
     }
   }
-}
+}

packages/nx/src/executors/atomizer/atomizer.impl.spec.ts

@@ -0,0 +1,74 @@
+import '../../internal-testing-utils/mock-fs';
+
+import { vol } from 'memfs';
+import { ExecutorContext } from '../../config/misc-interfaces';
+import atomize from './atomizer.impl';
+import * as fs from 'fs';
+import { join } from 'path';
+
+const context = {
+  root: '/root',
+  projectName: 'proj',
+  targetName: 'e2e-ci',
+} as ExecutorContext;
+
+describe('nx:atomizer', () => {
+  let mockProcessExit: jest.SpyInstance;
+  let env: NodeJS.ProcessEnv;
+
+  beforeEach(() => {
+    env = process.env;
+    process.env = {};
+
+    jest.mock('fs');
+  });
+
+  afterEach(() => {
+    process.env = env;
+    vol.reset();
+  });
+
+  it('should fail if atomized task is present but no DTE', async () => {
+    const result = await atomize({}, context);
+    expect(result).toEqual(expect.objectContaining({ success: false }));
+  });
+
+  it('should do nothing if atomized task is present in Nx Agents', async () => {
+    process.env['NX_AGENT_NAME'] = '123';
+    const result = await atomize({}, context);
+    expect(result).toEqual(expect.objectContaining({ success: true }));
+  });
+
+  it('should do nothing if atomized task is present in DTE', async () => {
+    process.env['NX_CLOUD_DISTRIBUTED_EXECUTION_ID'] = '123';
+    const result = await atomize({}, context);
+    expect(result).toEqual(expect.objectContaining({ success: true }));
+  });
+
+  it('should do nothing if atomized task is present and dte marker file exists', async () => {
+    vol.fromJSON(
+      {
+        'node_modules/.cache/nx/NX_CLOUD_DISTRIBUTED_EXECUTION': 'true',
+      },
+      context.root
+    );
+
+    const result = await atomize({}, context);
+    expect(result).toEqual(expect.objectContaining({ success: true }));
+  });
+
+  it('should do nothing if atomized task is present and dte marker file exists in NX_CACHE_DIRECTORY', async () => {
+    const cacheDirectory = join('node_modules', 'my-cache-dir');
+    process.env['NX_CACHE_DIRECTORY'] = cacheDirectory;
+
+    vol.fromJSON(
+      {
+        'node_modules/my-cache-dir/NX_CLOUD_DISTRIBUTED_EXECUTION': 'true',
+      },
+      context.root
+    );
+
+    const result = await atomize({}, context);
+    expect(result).toEqual(expect.objectContaining({ success: true }));
+  });
+});

packages/nx/src/executors/atomizer/atomizer.impl.ts

@@ -0,0 +1,45 @@
+import { join } from 'path';
+import { ExecutorContext } from '../../config/misc-interfaces';
+import { existsSync, readFileSync } from 'fs';
+import { output } from '../../utils/output';
+import { serializeTarget } from '../../utils/serialize-target';
+
+export default async function (_: any, context: ExecutorContext) {
+  if (isInDTE(context.root)) {
+    return { success: true };
+  } else {
+    output.error({
+      title: `The ${serializeTarget(
+        context.projectName,
+        context.targetName,
+        context.configurationName
+      )} task uses the atomizer and should only be run with Nx Cloud distribution.`,
+      bodyLines: [
+        'Learn more at https://nx.dev/ci/features/split-e2e-tasks#use-atomizer-only-with-nx-cloud-distribution',
+      ],
+    });
+    return { success: false };
+  }
+}
+
+function isInDTE(workspaceRoot: string): boolean {
+  if (
+    process.env['NX_CLOUD_DISTRIBUTED_EXECUTION_ID'] ||
+    process.env['NX_AGENT_NAME']
+  ) {
+    return true;
+  }
+
+  // checks for DTE marker file - needed so we can check for DTE on the main nx job
+  const nxCacheDirectory = process.env.NX_CACHE_DIRECTORY
+    ? [process.env['NX_CACHE_DIRECTORY']]
+    : ['node_modules', '.cache', 'nx'];
+  const dir = join(workspaceRoot, ...nxCacheDirectory);
+  const dteMarker = join(dir, 'NX_CLOUD_DISTRIBUTED_EXECUTION');
+
+  if (existsSync(dteMarker) && readFileSync(dteMarker, 'utf-8') === 'true') {
+    return true;
+  }
+
+  return false;
+}

packages/nx/src/executors/atomizer/schema.json

@@ -0,0 +1,10 @@
+{
+    "version": 2,
+    "title": "Atomizer",
+    "description": "An executor that is used as the root task for atomized tasks. Ensures it's being run in distribution and otherwise does nothing.",
+    "type": "object",
+    "cli": "nx",
+    "outputCapture": "pipe",
+    "properties": {},
+    "additionalProperties": true
+}

packages/nx/src/project-graph/utils/project-configuration-utils.ts

@@ -904,8 +904,12 @@ export function isCompatibleTarget(
 ) {
   const oneHasNoExecutor = !a.executor || !b.executor;
   const bothHaveSameExecutor = a.executor === b.executor;
+  const areNoopAndAtomizer =
+    (a.executor === 'nx:noop' && b.executor === 'nx:atomizer') ||
+    (a.executor === 'nx:atomizer' && b.executor === 'nx:noop');
 
   if (oneHasNoExecutor) return true;
+  if (areNoopAndAtomizer) return true;
   if (!bothHaveSameExecutor) return false;
 
   const isRunCommands = a.executor === 'nx:run-commands';

packages/nx/src/tasks-runner/run-command.ts

@@ -28,11 +28,7 @@ import { StoreRunInformationLifeCycle } from './life-cycles/store-run-informatio
 import { TaskHistoryLifeCycle } from './life-cycles/task-history-life-cycle';
 import { TaskProfilingLifeCycle } from './life-cycles/task-profiling-life-cycle';
 import { TaskTimingsLifeCycle } from './life-cycles/task-timings-life-cycle';
-import {
-  findCycle,
-  makeAcyclic,
-  validateAtomizedTasks,
-} from './task-graph-utils';
+import { findCycle, makeAcyclic } from './task-graph-utils';
 import { TasksRunner, TaskStatus } from './tasks-runner';
 import { shouldStreamOutput } from './utils';
 
@@ -95,7 +91,7 @@ async function getTerminalOutputLifeCycle(
   }
 }
 
-function createTaskGraphAndRunValidations(
+function createTaskGraphAndValidateCycles(
   projectGraph: ProjectGraph,
   extraTargetDependencies: TargetDependencies,
   projectNames: string[],
@@ -133,8 +129,6 @@ function createTaskGraphAndRunValidations(
     }
   }
 
-  validateAtomizedTasks(taskGraph, projectGraph);
-
   return taskGraph;
 }
 
@@ -153,7 +147,7 @@ export async function runCommand(
     async () => {
       const projectNames = projectsToRun.map((t) => t.name);
 
-      const taskGraph = createTaskGraphAndRunValidations(
+      const taskGraph = createTaskGraphAndValidateCycles(
         projectGraph,
         extraTargetDependencies ?? {},
         projectNames,