jason810496 commented on code in PR #76365: URL: https://github.com/apache/airflow/pull/76365#discussion_r4520557641
########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,201 @@ +<!-- + 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)). `inputs` is + the gate's own data, as any task's is. +3. **The terminal task is the body's single leaf**, which the SDK orders before the gate. Wiring the + leaf into `inputs` is what gets the decision checked against what the body produces. +4. **`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. +5. **`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. +6. **A body task returns a `LoopTaskRef<T>`**, a `TaskRef<T>` that records the loop it came from. + `TaskInputs` accepts any `TaskRef` for any argument, so `tsc` checks neither reading; the SDK + picks the reading from where the consumer sits. +7. **`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 cleanup = dag.task(async function cleanup() {}); + +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 6 is what lets one ref serve both. + +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. `TArgs` is its own arguments, one key each, exactly as a task handler takes + * its own. Nothing of the SDK's is in `TArgs`; the loop context comes from `getLoop()`. + */ +type UntilCallable<TArgs extends object> = ( + args: TArgs, +) => LoopDecision | boolean | Promise<LoopDecision | boolean>; + +interface TaskGroupRef { + loop(options: LoopOptions): LoopRef; + /** + * `NoInfer` keeps `TArgs` coming from the handler alone, so `inputs` is checked key by key against + * it rather than widening it. `dag.if` takes its inputs the same way. + */ + loop<TArgs extends object>( + until: UntilCallable<TArgs>, + inputs: TaskInputs<NoInfer<TArgs>>, + options: LoopOptions, + ): LoopRef; +} + +/** What `loop()` returns: the whole loop as one node, which is what an ordering edge leaves from. */ +interface LoopRef extends Node { + readonly groupId: string; + before(...downstream: readonly Node[]): LoopRef; + after(...upstream: readonly Node[]): LoopRef; +} + +/** + * What a loop body task returns. It is a `TaskRef<T>`, so it goes wherever one goes; the loop it + * records is what lets the SDK tell a body ref from an ordinary one when the Dag is read. + */ +interface LoopTaskRef<T> extends TaskRef<T> { + readonly loopId: string; +} + +/** + * Read inside a loop body or its gate. Throws anywhere else. + * Awaited because it reads the previous iteration's result, which it caches per task run. + */ +function getLoop(): Promise<LoopContext>; + +interface LoopContext { + readonly index: number; + readonly maxIterations: number; + + /** The previous iteration's terminal result as it arrived, or `null` at iteration 1. */ + readonly previous: unknown; + + /** The same value, as the type the caller names. `client.getXCom`'s type parameter. */ + getPrevious<T>(): T | null; +} +``` + +## Consequences + +- **A fixed-count loop needs no runtime support.** `loop({ maxIterations })` emits `has_until: false`, + which core completes without a worker (ADR-0017 decision 4). +- **The gate is an ordinary task.** `LoopOptions` extends `TaskSpec`, so `queue` and `retries` land on + it, and its id defaults to the handler's function name under the group's prefix. A fixed-count loop + has no handler to take a name from, so its gate is `__loop_gate`, as Python's is. +- **The body's single leaf is checked twice**: by the SDK when the Dag is read, for the faster + message, and by the Dag processor, which decides (ADR-0017 decision 3). A nested loop and a loop + inside a mapped task group are rejected the same way. +- **`getLoop()` throws outside a loop**, so a handler moved out of a group fails where it was moved. +- **A mapped terminal needs no new call.** `getPrevious<XComSequence<Metrics>>()` names the sequence + type, and `terminal_is_mapped` tells the runtime which to build. +- **The reading has to match where the consumer sits.** Core serves only the all-iterations read from Review Comment: Confirmed, and taken. `serializeArgBindings` records a task id and no argument type, so nothing could check which reading a consumer declared. `evaluated.all()` now returns `TaskRef<XComSequence<T>>`, and a bare body ref outside the loop is rejected when the Dag is read. Go and Java keep the bare ref, since registration and the generated overloads can check it; Alternatives records why TypeScript cannot. [`aadfcd5`](https://github.com/apache/airflow/pull/76365/commits/aadfcd5370e) ########## ts-sdk/adr/0003-task-loop-authoring-interface.md: ########## @@ -0,0 +1,201 @@ +<!-- + 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)). `inputs` is + the gate's own data, as any task's is. +3. **The terminal task is the body's single leaf**, which the SDK orders before the gate. Wiring the + leaf into `inputs` is what gets the decision checked against what the body produces. +4. **`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. +5. **`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. Review Comment: Right on both counts: `getXCom<T = unknown>()` is one method, and ADR-0017 asks every runtime for one accessor. The `previous` field is gone and `getPrevious<T = unknown>()` stands alone. [`f7a02f4`](https://github.com/apache/airflow/pull/76365/commits/f7a02f4a006) ########## go-sdk/adr/0009-task-loop-authoring-interface.md: ########## @@ -0,0 +1,245 @@ +<!-- + 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`**, which is its own data, as any task's + is. The terminal task is the body's single leaf, which the SDK orders before the gate. Wiring the + leaf in is what gets the decision checked against what the body produces, since registration + already checks a parameter against the recorded output type of whatever `Inputs` names. +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, + and outside a loop that error is `airflow.NotInLoop`. +5. **`Index`, `MaxIterations` and `Previous` are struct fields**, as `DagRun`'s are, and + `UnmarshalPrevious(out)` binds the raw one. That is the split `GetXCom` declares, whose `value` + parameter is reserved for exactly this decoding and ignored today. +6. **A body ref is read from outside the loop unchanged.** The consuming handler declares `T` for one + iteration or `airflow.XComSequence` for every iteration, and registration checks which it asked + for against where it sits. +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) error { + n, err := history.Len() + if err != nil { + return err + } + var last Metrics + if err := history.UnmarshalAt(n-1, &last); err != nil { + return err + } + actx.Logger().Info("settled", "iterations", n, "error", last.Error) + return nil +} + +func cleanup(actx airflow.Context) error { + 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`, 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. Outside a loop body or its gate it +// reports NotInLoop, and it reports the read's own error when that fails. +func (c Context) Loop() (Loop, error) + +// NotInLoop is what Loop reports for a handler that is not in a loop. Spelled like the SDK's other +// sentinels, sdk.XComNotFound among them: errors.Is(err, airflow.NotInLoop). +var NotInLoop = errors.New("not in a loop") + +// 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: Matched. [`832ae14`](https://github.com/apache/airflow/pull/76365/commits/832ae14dc2d) -- 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]
