TS SDK: reject a Dag whose order-only edges form a cycle

airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst

@@ -311,6 +311,33 @@ already exists changes nothing. Each returns the reference it was called on, so
 Pass a value as an argument when the downstream task needs it, and use ``before`` or ``after`` when
 it only needs to run in order.
 
+Task groups
+~~~~~~~~~~~
+
+``dag.taskGroup(groupId)`` opens a scope with the same ``task`` and ``taskGroup`` methods as the Dag,
+prefixing the id of everything declared in it, as Python's ``prefix_group_id`` does:
+
+.. code-block:: typescript
+
+    const staging = dag.taskGroup("staging");
+    staging.task("stage_rows", stageRows)();        // task id "staging.stage_rows"
+    staging.taskGroup("checks").task("nulls", checkNulls)();  // "staging.checks.nulls"
+
+    staging.before(loaded);                          // staging >> loaded
+
+A group is an edge endpoint in its own right, so ``before`` and ``after`` order a whole group against
+a task or against another group.
+
+Tasks and groups share one id namespace, as they do in Python, so a Dag cannot hold both a task and a
+group called ``staging``. A ``.`` is what separates a group from what it holds, so it cannot appear in
+an id of either.
+
+Pass ``{ prefixGroupId: false }`` to keep the ids declared in a group as written, as ``prefix_group_id=False``
+does in Python; they then have to be unique across the Dag. A group id is made of letters, digits, dashes and
+underscores, and is at most 200 characters.
+
+.. warning:: A cycle is not allowed. The Dag is rejected when it is read, naming the tasks on the cycle.
+
 ``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
 ``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task.
 

ts-sdk/src/index.ts

@@ -28,7 +28,10 @@ export type { ArgNameMap } from "./sdk/arg-names.js";
 export type { Registerable } from "./sdk/bundle.js";
 export type {
   DagSpec,
+  Node,
   TaskFactory,
+  TaskGroupOptions,
+  TaskGroupRef,
   TaskInput,
   TaskInputs,
   TaskOptions,

ts-sdk/src/sdk/cycle.ts

@@ -0,0 +1,89 @@
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Finds a cycle in a Dag's task graph.
+//
+// Only the edges `before`/`after` draw can form one. A cycle through *wiring*
+// is unrepresentable rather than rejected: a `TaskRef` exists only once its
+// producing call has returned, so there is no way to write one. Both kinds of
+// edge are searched together all the same, because a cycle can run through one
+// of each — `transform({ extracted })` and then `extract.after(transform)`.
+
+/** A directed edge between two task IDs. */
+export interface TaskEdge {
+  readonly upstream: string;
+  readonly downstream: string;
+}
+
+/**
+ * The tasks on a cycle, or undefined when there is none.
+ *
+ * The returned path starts and ends on the same task, so it reads as the cycle
+ * it describes: `["a", "b", "a"]` is `a >> b >> a`.
+ *
+ * Iterative rather than recursive: a Dag of a few thousand chained tasks is
+ * within reach for a generated Dag file, and would overflow the stack.
+ */
+export function findTaskCycle(
+  taskIds: Iterable<string>,
+  edges: Iterable<TaskEdge>,
+): string[] | undefined {
+  const downstreamOf = new Map<string, string[]>();
+  for (const taskId of taskIds) downstreamOf.set(taskId, []);
+  for (const { upstream, downstream } of edges) {
+    // An edge naming a task the Dag does not hold cannot be part of a cycle
+    // among the tasks it does, and the caller reports it in its own terms.
+    downstreamOf.get(upstream)?.push(downstream);
+  }
+
+  // Three states, as a depth-first cycle search needs: unvisited, on the
+  // current path, and finished. Finding a task already on the path is the
+  // cycle; finding a finished one only means the search has been there before.
+  const onPath = new Set<string>();
+  const finished = new Set<string>();
+  const path: string[] = [];
+
+  for (const root of downstreamOf.keys()) {
+    if (finished.has(root)) continue;
+    // Each frame is a task and how far its downstream list has been walked.
+    const stack: { taskId: string; next: number }[] = [{ taskId: root, next: 0 }];
+    onPath.add(root);
+    path.push(root);
+
+    while (stack.length > 0) {
+      const frame = stack[stack.length - 1]!;
+      const downstream = downstreamOf.get(frame.taskId) ?? [];
+      if (frame.next === downstream.length) {
+        finished.add(frame.taskId);
+        onPath.delete(frame.taskId);
+        path.pop();
+        stack.pop();
+        continue;
+      }
+      const next = downstream[frame.next]!;
+      frame.next += 1;
+      if (onPath.has(next)) return [...path.slice(path.indexOf(next)), next];
+      if (finished.has(next) || !downstreamOf.has(next)) continue;
+      onPath.add(next);
+      path.push(next);
+      stack.push({ taskId: next, next: 0 });
+    }
+  }
+  return undefined;
+}