jason810496 commented on code in PR #76365: URL: https://github.com/apache/airflow/pull/76365#discussion_r4509862369
########## go-sdk/adr/0009-task-loop-authoring-interface.md: ########## @@ -0,0 +1,211 @@ +<!-- + 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. + --> + +# 9. Task loop authoring interface + +Date: 2026-11-03 + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Go spelling of them. + +## Decision + +1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and fills it as usual + ([ADR 8](0008-native-dag-interface.md)). +2. **`maxIterations` is positional and `Until` is an option.** Every loop has a bound; only some add a + callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as `TaskOption` is. +3. **The gate is a task handler wired with `airflow.Inputs`.** Registration checks its parameter + against the terminal task's recorded output type, as it already does for `Inputs`. +4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, `TaskInstance` and + `DagRun`. It returns a value and an error, because reading the previous iteration is a round trip. +5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as `DagRun`'s are, and + `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split between the value it returns + and the `out` it decodes into. +6. **A body ref is read from outside the loop unchanged.** The consuming handler declares `T` for one + iteration or `airflow.XComSequence[T]` for every iteration, checked at registration. +7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a body ref to a task + outside is rejected when it is declared. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```go +type Metrics struct { + Estimate float64 `json:"estimate"` + Error float64 `json:"error"` +} + +func improve(actx airflow.Context) (float64, error) { + loop, err := actx.Loop() + if err != nil { + return 0, err + } + actx.Logger().Info("refining", "iteration", loop.Index, "of", loop.MaxIterations) + + estimate := 1.0 + if loop.Previous != nil { + var previous Metrics + if err := loop.UnmarshalPrevious(&previous); err != nil { + return 0, err + } + estimate = previous.Estimate + } + return (estimate + 2.0/estimate) / 2.0, nil +} + +func evaluate(actx airflow.Context, estimate float64) (Metrics, error) { + return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 2)}, nil +} + +func accurateEnough(actx airflow.Context, result Metrics) (airflow.LoopDecision, error) { + return airflow.LoopDecision{ + Stop: result.Error < 1e-6, + Reason: fmt.Sprintf("error %.2g", result.Error), + }, nil +} + +func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error { + n, err := history.Len() + if err != nil { + return err + } + actx.Logger().Info("settled", "iterations", n) + return nil +} +``` + +```go +dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"}) +refine := dag.TaskGroup("refine") + +improved := refine.Task(improve) +evaluated := refine.Task(evaluate, airflow.Inputs(improved)) + +refinement := refine.Loop(10, airflow.Until(accurateEnough, airflow.Inputs(evaluated))) + +dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, through the gate +refinement.Before(dag.Task(cleanup)) // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `airflow.XComSequence[Metrics]`, which is what +`report` takes. Decision 6 is what makes registration accept both. + +A loop that only counts needs no callable: + +```go +poll := dag.TaskGroup("poll") +poll.Task(checkOnce) +poll.Loop(3) +``` + +## Signature + +```go +package airflow + +// Loop turns the group into a task loop. maxIterations must be positive. +func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef + +// Until supplies the gate's callable and the gate's own task options, including the Inputs that +// name the loop's terminal task. fn takes airflow.Context first and the wired inputs after, as +// every handler does, and returns either shape: +// +// func(actx airflow.Context, args ...) (LoopDecision, error) +// func(actx airflow.Context, args ...) (bool, error) +func Until(fn any, opts ...TaskOption) LoopOption + +// LoopOption is sealed: its only method is unexported, so a loop takes SDK-defined options and +// nothing else. +type LoopOption interface{ applyLoop(*loopConfig) error } + +// LoopDecision is what an until callable returns. A bool return is a decision with no reason. +type LoopDecision struct { + Stop bool + Reason string +} + +// LoopRef is a Node, so Before and After carry edges for the whole loop. +type LoopRef struct{ /* unexported */ } + +// Loop reads the previous iteration's result once and caches it. It panics outside a loop body or +// gate, as the other accessors do on a zero Context, and returns an error when the read fails. +func (c Context) Loop() (Loop, error) + +// Loop is handed back by value, as DagRun is: the fields are the runtime's answer, not state for +// a handler to write. +type Loop struct { + Index int + MaxIterations int + + // Previous is the previous iteration's terminal result as it arrived: nil at Index 1, and an + // XComSequence when the terminal task is mapped. + Previous any +} + +// UnmarshalPrevious binds Previous to a DTO, the way XComClient.GetXCom decodes into its out +// parameter. It leaves out untouched when there is no previous iteration. +func (l Loop) UnmarshalPrevious(out any) error + +// XComSequence is the lazy sequence of ADR-0016: every iteration of a loop task read from outside +// the loop, or every instance of a mapped task. It is generic because it is reached as a declared +// parameter type, which the SDK reads by reflection as it does any other parameter. +type XComSequence[T any] interface { Review Comment: Yes. Reflection instantiates no generic type, so the runtime could never build an `XComSequence[Metrics]` to pass, nor assert `Previous` to it. Non-generic now, with `At(offset) (any, error)` and `UnmarshalAt(offset, out) error` beside `UnmarshalPrevious`. [`a97060b`](https://github.com/apache/airflow/commit/a97060b0483d58a252613bfe3053c89d68db0513) ########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,166 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — TypeScript Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the TypeScript spelling of them. + +## Decision + +1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and fills it as usual. +2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, the shape + `dag.if(handler, inputs, spec)` already has ([ADR-0002](0002-native-dag-interface.md)). +3. **`getLoop()` supplies the loop context**, beside `getContext()` and `getClient()` (ADR-0002 + decision 6). It is awaited, because reading the previous iteration is a round trip. +4. **`index`, `maxIterations` and `previous` are fields**, and `getPrevious<T>()` is the typed view. + That is `client.getXCom`'s split between a value and the type the caller names. +5. **A body task returns a `LoopTaskRef<T>`**, accepted where either `TaskRef<T>` or Review Comment: Yes, in all three: a single-value read from outside the loop is rejected when the Dag is read. [`7523b1f`](https://github.com/apache/airflow/commit/7523b1f2c3422b66bc77cdbfa5056043d294a978) ########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,166 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — TypeScript Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the TypeScript spelling of them. + +## Decision + +1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and fills it as usual. +2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, the shape + `dag.if(handler, inputs, spec)` already has ([ADR-0002](0002-native-dag-interface.md)). +3. **`getLoop()` supplies the loop context**, beside `getContext()` and `getClient()` (ADR-0002 + decision 6). It is awaited, because reading the previous iteration is a round trip. +4. **`index`, `maxIterations` and `previous` are fields**, and `getPrevious<T>()` is the typed view. + That is `client.getXCom`'s split between a value and the type the caller names. +5. **A body task returns a `LoopTaskRef<T>`**, accepted where either `TaskRef<T>` or + `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the consumer declares. +6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body ref to a task outside is + rejected when the Dag is read. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```ts +import { Dag, getLoop, type LoopDecision, type XComSequence } from "apache-airflow-ts-sdk"; + +type Metrics = { estimate: number; error: number }; + +const dag = new Dag("ts_refine", { schedule: "@daily" }); +const refine = dag.taskGroup("refine"); + +const improve = refine.task(async function improve(): Promise<number> { + const loop = await getLoop(); + console.log(`iteration ${loop.index} of ${loop.maxIterations}`); + const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0; + return (estimate + 2.0 / estimate) / 2.0; +}); + +const evaluate = refine.task(async function evaluate({ estimate }: { estimate: number }): Promise<Metrics> { + return { estimate, error: Math.abs(estimate * estimate - 2) }; +}); + +async function accurateEnough({ result }: { result: Metrics }): Promise<LoopDecision> { + return { stop: result.error < 1e-6, reason: `error ${result.error.toExponential(1)}` }; +} + +const report = dag.task(async function report({ history }: { history: XComSequence<Metrics> }) { + console.log(`settled after ${await history.length()} iterations`); +}); + +const estimate = improve(); +const evaluated = evaluate({ estimate }); + +const refinement = refine.loop(accurateEnough, { result: evaluated }, { maxIterations: 10 }); + +report({ history: evaluated }); // every iteration's metrics, through the gate +refinement.before(cleanup()); // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which is what `report` takes. +Decision 5 is what makes both type-check. + +A loop that only counts needs no handler: + +```ts +const poll = dag.taskGroup("poll"); +poll.task(checkOnce)(); +poll.loop({ maxIterations: 3 }); +``` + +## Signature + +```ts +/** Options of a loop, the trailing argument of `loop()`. */ +interface LoopOptions extends TaskSpec { + maxIterations: number; +} + +/** What `until` returns. A bare `boolean` is `{ stop }`. */ +interface LoopDecision { + stop: boolean; + reason?: string; +} + +/** + * The gate's handler. `A` is its wired inputs, one key per argument, exactly as a task handler + * takes its own. Nothing of the SDK's is in `A`; the loop context comes from `getLoop()`. + */ +type UntilCallable<A> = (args: A) => Promise<LoopDecision | boolean>; + +interface TaskGroupRef { + loop(options: LoopOptions): LoopRef; + loop<A>(until: UntilCallable<A>, inputs: A, options: LoopOptions): LoopRef; Review Comment: Yes, `dag.if`'s shape: `inputs: TaskInputs<NoInfer<TArgs>>`, plus `LoopRef` and `LoopTaskRef<T>` in the block. Decision 6 also overpromised: `TaskInputs` takes an unparameterized `TaskRef`, so `tsc` checks neither reading. It now says what `tsc` does check. [`bc7d0fd`](https://github.com/apache/airflow/commit/bc7d0fd2fc95a662d107d3e32e6c5b28de1f9f59) ########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,166 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — TypeScript Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the TypeScript spelling of them. + +## Decision + +1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and fills it as usual. +2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, the shape + `dag.if(handler, inputs, spec)` already has ([ADR-0002](0002-native-dag-interface.md)). +3. **`getLoop()` supplies the loop context**, beside `getContext()` and `getClient()` (ADR-0002 + decision 6). It is awaited, because reading the previous iteration is a round trip. +4. **`index`, `maxIterations` and `previous` are fields**, and `getPrevious<T>()` is the typed view. + That is `client.getXCom`'s split between a value and the type the caller names. +5. **A body task returns a `LoopTaskRef<T>`**, accepted where either `TaskRef<T>` or + `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the consumer declares. +6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body ref to a task outside is + rejected when the Dag is read. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```ts +import { Dag, getLoop, type LoopDecision, type XComSequence } from "apache-airflow-ts-sdk"; + +type Metrics = { estimate: number; error: number }; + +const dag = new Dag("ts_refine", { schedule: "@daily" }); +const refine = dag.taskGroup("refine"); + +const improve = refine.task(async function improve(): Promise<number> { + const loop = await getLoop(); + console.log(`iteration ${loop.index} of ${loop.maxIterations}`); + const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0; + return (estimate + 2.0 / estimate) / 2.0; +}); + +const evaluate = refine.task(async function evaluate({ estimate }: { estimate: number }): Promise<Metrics> { + return { estimate, error: Math.abs(estimate * estimate - 2) }; +}); + +async function accurateEnough({ result }: { result: Metrics }): Promise<LoopDecision> { + return { stop: result.error < 1e-6, reason: `error ${result.error.toExponential(1)}` }; +} + +const report = dag.task(async function report({ history }: { history: XComSequence<Metrics> }) { + console.log(`settled after ${await history.length()} iterations`); +}); + +const estimate = improve(); +const evaluated = evaluate({ estimate }); + +const refinement = refine.loop(accurateEnough, { result: evaluated }, { maxIterations: 10 }); + +report({ history: evaluated }); // every iteration's metrics, through the gate +refinement.before(cleanup()); // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which is what `report` takes. +Decision 5 is what makes both type-check. + +A loop that only counts needs no handler: + +```ts +const poll = dag.taskGroup("poll"); +poll.task(checkOnce)(); +poll.loop({ maxIterations: 3 }); Review Comment: Yes, `__loop_gate` in all three, matching `_internal/loop.py:205-207`. [`bc98ba3`](https://github.com/apache/airflow/commit/bc98ba3d0c262d9258ddf79b5d2ea27c193f164a) ########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,166 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — TypeScript Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the TypeScript spelling of them. + +## Decision + +1. **`refine.loop(...)` closes a group.** `dag.taskGroup("refine")` builds and fills it as usual. +2. **The gate is a handler wired in one call**: `loop(until, inputs, spec)`, the shape + `dag.if(handler, inputs, spec)` already has ([ADR-0002](0002-native-dag-interface.md)). +3. **`getLoop()` supplies the loop context**, beside `getContext()` and `getClient()` (ADR-0002 + decision 6). It is awaited, because reading the previous iteration is a round trip. +4. **`index`, `maxIterations` and `previous` are fields**, and `getPrevious<T>()` is the typed view. + That is `client.getXCom`'s split between a value and the type the caller names. +5. **A body task returns a `LoopTaskRef<T>`**, accepted where either `TaskRef<T>` or + `TaskRef<XComSequence<T>>` is wanted, so `tsc` checks whichever reading the consumer declares. +6. **`LoopRef` carries `before` and `after`.** An order-only edge from a body ref to a task outside is + rejected when the Dag is read. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```ts +import { Dag, getLoop, type LoopDecision, type XComSequence } from "apache-airflow-ts-sdk"; + +type Metrics = { estimate: number; error: number }; + +const dag = new Dag("ts_refine", { schedule: "@daily" }); +const refine = dag.taskGroup("refine"); + +const improve = refine.task(async function improve(): Promise<number> { + const loop = await getLoop(); + console.log(`iteration ${loop.index} of ${loop.maxIterations}`); + const estimate = loop.getPrevious<Metrics>()?.estimate ?? 1.0; + return (estimate + 2.0 / estimate) / 2.0; +}); + +const evaluate = refine.task(async function evaluate({ estimate }: { estimate: number }): Promise<Metrics> { + return { estimate, error: Math.abs(estimate * estimate - 2) }; +}); + +async function accurateEnough({ result }: { result: Metrics }): Promise<LoopDecision> { + return { stop: result.error < 1e-6, reason: `error ${result.error.toExponential(1)}` }; +} + +const report = dag.task(async function report({ history }: { history: XComSequence<Metrics> }) { + console.log(`settled after ${await history.length()} iterations`); +}); + +const estimate = improve(); +const evaluated = evaluate({ estimate }); + +const refinement = refine.loop(accurateEnough, { result: evaluated }, { maxIterations: 10 }); + +report({ history: evaluated }); // every iteration's metrics, through the gate +refinement.before(cleanup()); // order only: the whole loop, not one iteration Review Comment: Declared in both. [`3be5b7b`](https://github.com/apache/airflow/commit/3be5b7bcf9ebdfdd0cfbbce78db4e65c555df6f1) ########## go-sdk/adr/0009-task-loop-authoring-interface.md: ########## @@ -0,0 +1,211 @@ +<!-- + 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. + --> + +# 9. Task loop authoring interface + +Date: 2026-11-03 + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Go spelling of them. + +## Decision + +1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and fills it as usual + ([ADR 8](0008-native-dag-interface.md)). +2. **`maxIterations` is positional and `Until` is an option.** Every loop has a bound; only some add a + callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as `TaskOption` is. +3. **The gate is a task handler wired with `airflow.Inputs`.** Registration checks its parameter + against the terminal task's recorded output type, as it already does for `Inputs`. +4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, `TaskInstance` and + `DagRun`. It returns a value and an error, because reading the previous iteration is a round trip. +5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as `DagRun`'s are, and + `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split between the value it returns + and the `out` it decodes into. +6. **A body ref is read from outside the loop unchanged.** The consuming handler declares `T` for one + iteration or `airflow.XComSequence[T]` for every iteration, checked at registration. +7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a body ref to a task + outside is rejected when it is declared. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```go +type Metrics struct { + Estimate float64 `json:"estimate"` + Error float64 `json:"error"` +} + +func improve(actx airflow.Context) (float64, error) { + loop, err := actx.Loop() + if err != nil { + return 0, err + } + actx.Logger().Info("refining", "iteration", loop.Index, "of", loop.MaxIterations) + + estimate := 1.0 + if loop.Previous != nil { + var previous Metrics + if err := loop.UnmarshalPrevious(&previous); err != nil { + return 0, err + } + estimate = previous.Estimate + } + return (estimate + 2.0/estimate) / 2.0, nil +} + +func evaluate(actx airflow.Context, estimate float64) (Metrics, error) { + return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 2)}, nil +} + +func accurateEnough(actx airflow.Context, result Metrics) (airflow.LoopDecision, error) { + return airflow.LoopDecision{ + Stop: result.Error < 1e-6, + Reason: fmt.Sprintf("error %.2g", result.Error), + }, nil +} + +func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error { + n, err := history.Len() + if err != nil { + return err + } + actx.Logger().Info("settled", "iterations", n) + return nil +} +``` + +```go +dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"}) +refine := dag.TaskGroup("refine") + +improved := refine.Task(improve) +evaluated := refine.Task(evaluate, airflow.Inputs(improved)) + +refinement := refine.Loop(10, airflow.Until(accurateEnough, airflow.Inputs(evaluated))) + +dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, through the gate +refinement.Before(dag.Task(cleanup)) // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `airflow.XComSequence[Metrics]`, which is what +`report` takes. Decision 6 is what makes registration accept both. + +A loop that only counts needs no callable: + +```go +poll := dag.TaskGroup("poll") +poll.Task(checkOnce) +poll.Loop(3) +``` + +## Signature + +```go +package airflow + +// Loop turns the group into a task loop. maxIterations must be positive. +func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef + +// Until supplies the gate's callable and the gate's own task options, including the Inputs that Review Comment: Yes. The terminal is the body's single leaf, the SDK orders it before the gate, and the gate's `Inputs` is ordinary wiring. [`bc98ba3`](https://github.com/apache/airflow/commit/bc98ba3d0c262d9258ddf79b5d2ea27c193f164a), with rule 3 amended in #76364: [`d0c50c6`](https://github.com/apache/airflow/commit/d0c50c61fee49612f5931d293045d8938f946247) ########## go-sdk/adr/0009-task-loop-authoring-interface.md: ########## @@ -0,0 +1,211 @@ +<!-- + 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. + --> + +# 9. Task loop authoring interface + +Date: 2026-11-03 + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Go spelling of them. + +## Decision + +1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and fills it as usual + ([ADR 8](0008-native-dag-interface.md)). +2. **`maxIterations` is positional and `Until` is an option.** Every loop has a bound; only some add a + callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as `TaskOption` is. +3. **The gate is a task handler wired with `airflow.Inputs`.** Registration checks its parameter + against the terminal task's recorded output type, as it already does for `Inputs`. +4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, `TaskInstance` and + `DagRun`. It returns a value and an error, because reading the previous iteration is a round trip. +5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as `DagRun`'s are, and + `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split between the value it returns + and the `out` it decodes into. +6. **A body ref is read from outside the loop unchanged.** The consuming handler declares `T` for one + iteration or `airflow.XComSequence[T]` for every iteration, checked at registration. +7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a body ref to a task + outside is rejected when it is declared. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```go +type Metrics struct { + Estimate float64 `json:"estimate"` + Error float64 `json:"error"` +} + +func improve(actx airflow.Context) (float64, error) { + loop, err := actx.Loop() + if err != nil { + return 0, err + } + actx.Logger().Info("refining", "iteration", loop.Index, "of", loop.MaxIterations) + + estimate := 1.0 + if loop.Previous != nil { + var previous Metrics + if err := loop.UnmarshalPrevious(&previous); err != nil { + return 0, err + } + estimate = previous.Estimate + } + return (estimate + 2.0/estimate) / 2.0, nil +} + +func evaluate(actx airflow.Context, estimate float64) (Metrics, error) { + return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 2)}, nil +} + +func accurateEnough(actx airflow.Context, result Metrics) (airflow.LoopDecision, error) { + return airflow.LoopDecision{ + Stop: result.Error < 1e-6, + Reason: fmt.Sprintf("error %.2g", result.Error), + }, nil +} + +func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error { + n, err := history.Len() + if err != nil { + return err + } + actx.Logger().Info("settled", "iterations", n) + return nil +} +``` + +```go +dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"}) +refine := dag.TaskGroup("refine") + +improved := refine.Task(improve) +evaluated := refine.Task(evaluate, airflow.Inputs(improved)) + +refinement := refine.Loop(10, airflow.Until(accurateEnough, airflow.Inputs(evaluated))) + +dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, through the gate +refinement.Before(dag.Task(cleanup)) // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `airflow.XComSequence[Metrics]`, which is what +`report` takes. Decision 6 is what makes registration accept both. + +A loop that only counts needs no callable: + +```go +poll := dag.TaskGroup("poll") +poll.Task(checkOnce) +poll.Loop(3) +``` + +## Signature + +```go +package airflow + +// Loop turns the group into a task loop. maxIterations must be positive. +func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef + +// Until supplies the gate's callable and the gate's own task options, including the Inputs that +// name the loop's terminal task. fn takes airflow.Context first and the wired inputs after, as +// every handler does, and returns either shape: +// +// func(actx airflow.Context, args ...) (LoopDecision, error) +// func(actx airflow.Context, args ...) (bool, error) +func Until(fn any, opts ...TaskOption) LoopOption + +// LoopOption is sealed: its only method is unexported, so a loop takes SDK-defined options and +// nothing else. +type LoopOption interface{ applyLoop(*loopConfig) error } + +// LoopDecision is what an until callable returns. A bool return is a decision with no reason. +type LoopDecision struct { + Stop bool + Reason string +} + +// LoopRef is a Node, so Before and After carry edges for the whole loop. +type LoopRef struct{ /* unexported */ } + +// Loop reads the previous iteration's result once and caches it. It panics outside a loop body or Review Comment: Yes, `airflow.NotInLoop` — no `Err` prefix, matching `sdk.XComNotFound`. [`f256f04`](https://github.com/apache/airflow/commit/f256f049bd17e2840e067b90ddd5ec2ce0129f7c) ########## go-sdk/adr/0009-task-loop-authoring-interface.md: ########## @@ -0,0 +1,211 @@ +<!-- + 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. + --> + +# 9. Task loop authoring interface + +Date: 2026-11-03 + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Go spelling of them. + +## Decision + +1. **`refine.Loop(...)` closes a group.** `dag.TaskGroup("refine")` builds and fills it as usual + ([ADR 8](0008-native-dag-interface.md)). +2. **`maxIterations` is positional and `Until` is an option.** Every loop has a bound; only some add a + callable. `airflow.Until(fn, opts...)` is the one `LoopOption`, sealed as `TaskOption` is. +3. **The gate is a task handler wired with `airflow.Inputs`.** Registration checks its parameter + against the terminal task's recorded output type, as it already does for `Inputs`. +4. **`actx.Loop()` supplies the loop context**, beside `Logger`, `Client`, `TaskInstance` and + `DagRun`. It returns a value and an error, because reading the previous iteration is a round trip. +5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as `DagRun`'s are, and + `UnmarshalPrevious(out)` binds the raw one. That is `GetXCom`'s split between the value it returns + and the `out` it decodes into. +6. **A body ref is read from outside the loop unchanged.** The consuming handler declares `T` for one + iteration or `airflow.XComSequence[T]` for every iteration, checked at registration. +7. **`*LoopRef` is the `Node` that carries edges.** An order-only edge from a body ref to a task + outside is rejected when it is declared. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```go +type Metrics struct { + Estimate float64 `json:"estimate"` + Error float64 `json:"error"` +} + +func improve(actx airflow.Context) (float64, error) { + loop, err := actx.Loop() + if err != nil { + return 0, err + } + actx.Logger().Info("refining", "iteration", loop.Index, "of", loop.MaxIterations) + + estimate := 1.0 + if loop.Previous != nil { + var previous Metrics + if err := loop.UnmarshalPrevious(&previous); err != nil { + return 0, err + } + estimate = previous.Estimate + } + return (estimate + 2.0/estimate) / 2.0, nil +} + +func evaluate(actx airflow.Context, estimate float64) (Metrics, error) { + return Metrics{Estimate: estimate, Error: math.Abs(estimate*estimate - 2)}, nil +} + +func accurateEnough(actx airflow.Context, result Metrics) (airflow.LoopDecision, error) { + return airflow.LoopDecision{ + Stop: result.Error < 1e-6, + Reason: fmt.Sprintf("error %.2g", result.Error), + }, nil +} + +func report(actx airflow.Context, history airflow.XComSequence[Metrics]) error { + n, err := history.Len() + if err != nil { + return err + } + actx.Logger().Info("settled", "iterations", n) + return nil +} +``` + +```go +dag := airflow.Dag("go_refine", airflow.DagSpec{Schedule: "@daily"}) +refine := dag.TaskGroup("refine") + +improved := refine.Task(improve) +evaluated := refine.Task(evaluate, airflow.Inputs(improved)) + +refinement := refine.Loop(10, airflow.Until(accurateEnough, airflow.Inputs(evaluated))) + +dag.Task(report, airflow.Inputs(evaluated)) // every iteration's metrics, through the gate +refinement.Before(dag.Task(cleanup)) // order only: the whole loop, not one iteration +``` + +`evaluated` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `airflow.XComSequence[Metrics]`, which is what +`report` takes. Decision 6 is what makes registration accept both. + +A loop that only counts needs no callable: + +```go +poll := dag.TaskGroup("poll") +poll.Task(checkOnce) +poll.Loop(3) +``` + +## Signature + +```go +package airflow + +// Loop turns the group into a task loop. maxIterations must be positive. +func (g *TaskGroupRef) Loop(maxIterations int, opts ...LoopOption) *LoopRef + +// Until supplies the gate's callable and the gate's own task options, including the Inputs that +// name the loop's terminal task. fn takes airflow.Context first and the wired inputs after, as +// every handler does, and returns either shape: +// +// func(actx airflow.Context, args ...) (LoopDecision, error) +// func(actx airflow.Context, args ...) (bool, error) +func Until(fn any, opts ...TaskOption) LoopOption + +// LoopOption is sealed: its only method is unexported, so a loop takes SDK-defined options and +// nothing else. +type LoopOption interface{ applyLoop(*loopConfig) error } + +// LoopDecision is what an until callable returns. A bool return is a decision with no reason. +type LoopDecision struct { + Stop bool + Reason string +} + +// LoopRef is a Node, so Before and After carry edges for the whole loop. +type LoopRef struct{ /* unexported */ } + +// Loop reads the previous iteration's result once and caches it. It panics outside a loop body or +// gate, as the other accessors do on a zero Context, and returns an error when the read fails. +func (c Context) Loop() (Loop, error) + +// Loop is handed back by value, as DagRun is: the fields are the runtime's answer, not state for +// a handler to write. +type Loop struct { + Index int + MaxIterations int + + // Previous is the previous iteration's terminal result as it arrived: nil at Index 1, and an + // XComSequence when the terminal task is mapped. + Previous any +} + +// UnmarshalPrevious binds Previous to a DTO, the way XComClient.GetXCom decodes into its out Review Comment: Right. Decision 5 now cites it as the split `GetXCom` declares and that parameter is reserved for, not as something it does today. [`a97060b`](https://github.com/apache/airflow/commit/a97060b0483d58a252613bfe3053c89d68db0513) ########## java-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,276 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — Java Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Java spelling of them, on both surfaces +[ADR-0002](0002-native-dag-interface.md) defines. + +## Decision + +1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface surface, + `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` does. +2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence is a fixed-count loop. +3. **The gate's data parameter is the terminal task**, bound in `depends()` like any other argument. + The interface surface has no typed refs, so `refine.until(id, class)` takes the body's single leaf. +4. **`Loop` is injected as a method argument**, beside `Client` and `Context` (ADR-0002), into a body + task or the gate. On the interface surface, where `Task.execute(Context, Client)` is fixed, it is + `context.loop()`. The processor injects it only where it is declared, so nothing else pays for the + read. +5. **`index`, `maxIterations` and `previous` are public final fields**, as `Context.conf` and + `context.runId` already are. `previous` is `null` at iteration 1. +6. **`Loop<T>` is generic over the terminal task's result**, so `Loop<Metrics>` gives a typed + `previous` with nothing to cast. The interface surface has no declaration to read, so + `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names the type at the read, + as `Client.getXCom` does. +7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor generates every wiring + parameter of type `XComSequence<E>` as an overloaded pair, one taking `Arg<XComSequence<E>>` and + one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the consumer declares. +8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body ref to a task outside is + rejected when the Dag is read. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```java [email protected](id = "java_refine", schedule = "@daily") +public class RefineEstimate { + private static final System.Logger log = System.getLogger(RefineEstimate.class.getName()); + + record Metrics(double estimate, double error) {} + + // A loop body: a task group with a bound. "refine.improve", "refine.evaluate", + // "refine.accurate_enough". + @Airflow.Loop(id = "refine", maxIterations = 10) + static class Refine { + + @Airflow.Task(id = "improve") + public double improve(Loop<Metrics> loop) { + log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations); + double estimate = loop.previous == null ? 1.0 : loop.previous.estimate(); + return (estimate + 2.0 / estimate) / 2.0; + } + + @Airflow.Task(id = "evaluate") + public Metrics evaluate(double estimate) { + return new Metrics(estimate, Math.abs(estimate * estimate - 2)); + } + + // The gate. Its parameter is the loop's terminal task, bound in depends(). + @Airflow.Until(id = "accurate_enough") + public LoopDecision accurateEnough(Metrics result) { + return new LoopDecision(result.error() < 1e-6, "error %.2g".formatted(result.error())); + } + } + + @Airflow.Task(id = "report") + public void report(XComSequence<Metrics> history) { + log.log(INFO, "settled after {0} iterations", history.length()); + } + + @Airflow.Task(id = "cleanup") + public void cleanup() {} + + @Airflow.Deps + static class Wiring implements RefineEstimateDeps { + void depends() { + var estimate = refine().improve(); + var metrics = refine().evaluate(estimate); + refine().accurateEnough(metrics); // the gate, and the terminal it reads + report(metrics); // every iteration's metrics, through the gate + refine().before(cleanup()); // order only: the whole loop, not one iteration + } + } +} +``` + +`metrics` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which is what `report` takes. +Decision 7 is what makes both compile: + +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +interface RefineEstimateDeps extends Deps { + default Refine refine() { return Flow.loop("refine"); } + + interface Refine extends LoopRef { + default LoopTaskRef<Double> improve() { return Flow.node("refine.improve"); } + default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return Flow.call("refine.evaluate", estimate); } + default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return Flow.call("refine.accurate_enough", result); } + } + + default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return Flow.call("report", history); } + default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return Flow.call("report", history); } + default TaskRef<Void> cleanup() { return Flow.node("cleanup"); } +} +``` + +A loop that only counts needs no gate method at all: + +```java [email protected](id = "poll", maxIterations = 3) +static class Poll { + @Airflow.Task(id = "check_once") + public void checkOnce() {} +} +``` + +### Interface surface + +```java +Dag dag = new Dag("java_refine").config("schedule", "@daily"); + +LoopRef refine = dag.loop("refine", 10); +TaskRef improve = refine.task("improve", Improve.class); +TaskRef evaluate = refine.task("evaluate", Evaluate.class); +improve.before(evaluate); +refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's single leaf + +TaskRef report = dag.task("report", Report.class); +evaluate.before(report); // data edge out of the body, through the gate Review Comment: Both fixed: `refine.before(report)`, and `Report` reads `client.getXComSequence(Metrics.class, "refine.evaluate")`. A consequence now says this surface wires no data out of the body at all. [`dd56571`](https://github.com/apache/airflow/commit/dd5657138e32d1127e2f76a18f1a79e9c37f990c) ########## java-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,276 @@ +<!-- + 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. + --> + +# ADR-0003: Task Loop — Java Authoring Interface + +## Status + +Proposed. + +## Context + +A task loop runs a task group again and again until a gate says stop. What a loop means, and the +eight authoring rules every SDK follows, are in +[`airflow-core/adr/lang-sdk/0017`](../../airflow-core/adr/lang-sdk/0017-task-loops-across-lang-sdks.md). +This ADR records the Java spelling of them, on both surfaces +[ADR-0002](0002-native-dag-interface.md) defines. + +## Decision + +1. **`@Airflow.Loop` is `@Airflow.TaskGroup` with a bound.** On the interface surface, + `dag.loop(groupId, maxIterations)` stands where `dag.taskGroup(groupId)` does. +2. **`@Airflow.Until` marks the gate's method** inside that class. Its absence is a fixed-count loop. +3. **The gate's data parameter is the terminal task**, bound in `depends()` like any other argument. + The interface surface has no typed refs, so `refine.until(id, class)` takes the body's single leaf. +4. **`Loop` is injected as a method argument**, beside `Client` and `Context` (ADR-0002), into a body + task or the gate. On the interface surface, where `Task.execute(Context, Client)` is fixed, it is + `context.loop()`. The processor injects it only where it is declared, so nothing else pays for the + read. +5. **`index`, `maxIterations` and `previous` are public final fields**, as `Context.conf` and + `context.runId` already are. `previous` is `null` at iteration 1. +6. **`Loop<T>` is generic over the terminal task's result**, so `Loop<Metrics>` gives a typed + `previous` with nothing to cast. The interface surface has no declaration to read, so + `context.loop()` is a `Loop<Object>` and `getPrevious(Metrics.class)` names the type at the read, + as `Client.getXCom` does. +7. **A body task's wiring method returns `LoopTaskRef<T>`.** The processor generates every wiring + parameter of type `XComSequence<E>` as an overloaded pair, one taking `Arg<XComSequence<E>>` and + one taking `LoopTaskRef<E>`, so `javac` checks whichever reading the consumer declares. +8. **`LoopRef` carries `before` and `after`.** An order-only edge from a body ref to a task outside is + rejected when the Dag is read. + +## Example + +Newton's method for the square root of two, stopping on the error or at ten iterations. + +```java [email protected](id = "java_refine", schedule = "@daily") +public class RefineEstimate { + private static final System.Logger log = System.getLogger(RefineEstimate.class.getName()); + + record Metrics(double estimate, double error) {} + + // A loop body: a task group with a bound. "refine.improve", "refine.evaluate", + // "refine.accurate_enough". + @Airflow.Loop(id = "refine", maxIterations = 10) + static class Refine { + + @Airflow.Task(id = "improve") + public double improve(Loop<Metrics> loop) { + log.log(INFO, "iteration {0} of {1}", loop.index, loop.maxIterations); + double estimate = loop.previous == null ? 1.0 : loop.previous.estimate(); + return (estimate + 2.0 / estimate) / 2.0; + } + + @Airflow.Task(id = "evaluate") + public Metrics evaluate(double estimate) { + return new Metrics(estimate, Math.abs(estimate * estimate - 2)); + } + + // The gate. Its parameter is the loop's terminal task, bound in depends(). + @Airflow.Until(id = "accurate_enough") + public LoopDecision accurateEnough(Metrics result) { + return new LoopDecision(result.error() < 1e-6, "error %.2g".formatted(result.error())); + } + } + + @Airflow.Task(id = "report") + public void report(XComSequence<Metrics> history) { + log.log(INFO, "settled after {0} iterations", history.length()); + } + + @Airflow.Task(id = "cleanup") + public void cleanup() {} + + @Airflow.Deps + static class Wiring implements RefineEstimateDeps { + void depends() { + var estimate = refine().improve(); + var metrics = refine().evaluate(estimate); + refine().accurateEnough(metrics); // the gate, and the terminal it reads + report(metrics); // every iteration's metrics, through the gate + refine().before(cleanup()); // order only: the whole loop, not one iteration + } + } +} +``` + +`metrics` is written twice and means two things. Inside the loop it is one iteration's `Metrics`, +which is what the gate takes. Outside it is an `XComSequence<Metrics>`, which is what `report` takes. +Decision 7 is what makes both compile: + +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +interface RefineEstimateDeps extends Deps { + default Refine refine() { return Flow.loop("refine"); } + + interface Refine extends LoopRef { + default LoopTaskRef<Double> improve() { return Flow.node("refine.improve"); } + default LoopTaskRef<Metrics> evaluate(Arg<Double> estimate) { return Flow.call("refine.evaluate", estimate); } + default TaskRef<Void> accurateEnough(Arg<Metrics> result) { return Flow.call("refine.accurate_enough", result); } + } + + default TaskRef<Void> report(Arg<XComSequence<Metrics>> history) { return Flow.call("report", history); } + default TaskRef<Void> report(LoopTaskRef<Metrics> history) { return Flow.call("report", history); } + default TaskRef<Void> cleanup() { return Flow.node("cleanup"); } +} +``` + +A loop that only counts needs no gate method at all: + +```java [email protected](id = "poll", maxIterations = 3) +static class Poll { + @Airflow.Task(id = "check_once") + public void checkOnce() {} +} +``` + +### Interface surface + +```java +Dag dag = new Dag("java_refine").config("schedule", "@daily"); + +LoopRef refine = dag.loop("refine", 10); +TaskRef improve = refine.task("improve", Improve.class); +TaskRef evaluate = refine.task("evaluate", Evaluate.class); +improve.before(evaluate); +refine.until("accurate_enough", AccurateEnough.class); // terminal: the body's single leaf + +TaskRef report = dag.task("report", Report.class); +evaluate.before(report); // data edge out of the body, through the gate +refine.before(dag.task("cleanup", Cleanup.class)); +``` + +```java +public static class Improve implements Task { + @Override + public void execute(Context context, Client client) { + Metrics previous = context.loop().getPrevious(Metrics.class); + double estimate = previous == null ? 1.0 : previous.estimate(); + client.setXCom((estimate + 2.0 / estimate) / 2.0); + } +} + +public static class AccurateEnough implements UntilTask { + @Override + public LoopDecision decide(Context context, Client client) { + Metrics result = context.loop().getResult(Metrics.class); + return new LoopDecision(result.error() < 1e-6, "error %.2g".formatted(result.error())); + } +} +``` + +`UntilTask` joins `ConditionTask` and `SwitchTask` as a task kind with its own verb, so the gate is +recognized by the interface it implements rather than by a naming rule. + +## Signature + +```java +// Annotation surface. @Airflow.Until carries @Airflow.Task's own attributes as well. +@interface Airflow.Loop { String id(); int maxIterations(); } +@interface Airflow.Until { String id() default ""; } + +// The gate's method under @Airflow.Until. Its injected arguments come first and its wired data +// after, as any task method's do, and it returns either shape; a boolean is a decision with no +// reason. +// +// LoopDecision accurateEnough(Metrics result) +// LoopDecision accurateEnough(Loop<Metrics> loop, Metrics result) +// boolean accurateEnough(Metrics result) + +// Injected into a loop body task or its gate, beside Client and Context. On the interface +// surface it is context.loop(). It throws outside a loop. +// T is the terminal task's result type. Declare it (Loop<Metrics>) and previous is typed; +// declare none (Loop<Object>, which context.loop() returns) and name the type at the read. +public final class Loop<T> { + public final int index; + + public final int maxIterations; + + /** The previous iteration's terminal result, or null at iteration 1. */ + public final T previous; + + /** + * This iteration's terminal result, for a gate on the interface surface, where there is no + * ref to bind it from. Null in a body task. The annotation surface takes it as a parameter. + */ + public final T result; + + /** previous bound to a DTO, as Client.getXCom's caller names the type it wants. */ + public <R> R getPrevious(Class<R> type); + + public <R> R getResult(Class<R> type); +} + +/** The lazy sequence of ADR-0016, reached as a declared parameter type or as Loop's T. */ +public interface XComSequence<E> { + int length(); + + E get(int offset); +} + +/** What an until method returns. A boolean return is a decision with no reason. */ +public record LoopDecision(boolean stop, String reason) {} + +/** Interface surface. */ +public interface UntilTask { + LoopDecision decide(Context context, Client client); +} + +public interface LoopRef extends TaskGroupRef, Flow { Review Comment: Confirmed. Both become `open`, named in Consequences; constructors stay internal and `Arg` stays sealed. [`2456f60`](https://github.com/apache/airflow/commit/2456f606f6caa6f269c43a29657b08c34c3a214c) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
