Go-SDK: Dag/Task specs and downstream wiring in the authoring API#67155
Draft
jason810496 wants to merge 9 commits into
Draft
Go-SDK: Dag/Task specs and downstream wiring in the authoring API#67155jason810496 wants to merge 9 commits into
jason810496 wants to merge 9 commits into
Conversation
1 task
Closed
1 task
jason810496
added a commit
to jason810496/airflow
that referenced
this pull request
May 27, 2026
…dings - Omit map_index from SetXCom for unmapped task instances and carry it as *int from api.TaskInstance through to SetXComMsg. Fixes a compile-time mismatch and matches the supervisor's "absent vs -1" semantics for unmapped tasks. - Honor context cancellation in CoordinatorComm.Communicate by running the request send in a goroutine and selecting on ctx.Done(), so a blocked supervisor socket no longer wedges the caller. The underlying connection is left alone on cancel to avoid poisoning future writes with a stale deadline or leaving a partial length-prefixed frame on the wire. - Decode ConnectionResult.Login and Password as *string via a new mapStringPtr helper so an explicitly empty credential round-trips through the coordinator client instead of being silently treated as absent. sdk.Connection already encodes the distinction. Adds regression tests for each case. go-sdk: drop unused details field from CoordinatorClient CoordinatorClient stored a *StartupDetails it never read, and the companion PRs (apache#67155, apache#67318) that also construct or pass details into NewCoordinatorClient never read c.details either — they work directly off the local *StartupDetails inside RunTask. Keeping the field as API surface implies a contract the type does not honor, so remove it and let callers pass only the comm channel. go-sdk: unify ErrorResponse decoding in CoordinatorComm.Communicate The dispatcher response could carry an error in two places — the third element of a 3-tuple response frame, or as the body of a 2-tuple frame whose "type" is "ErrorResponse" — and Communicate inspected each path independently. The two branches diverged on nil-guarding decodeErrorResponse, which was easy-to-miss latent inconsistency: either path could grow a bug the other did not. Extract the source selection into errMapFromFrame so the decode and *ApiError construction live in one place. go-sdk: widen frame IDs to int64 end-to-end CoordinatorComm's request-id counter was atomic.Int64 (chosen so a long-running runtime cannot wrap), but the value was narrowed to int when stored in the pending map and IncomingFrame.ID. On 32-bit GOARCH that narrowing reintroduces the wraparound the int64 counter was meant to prevent, and the comment promising "wide enough to avoid wraparound" becomes architecture-dependent. Widen IncomingFrame.ID, the pending map key, encodeRequest, and the test fixtures to int64 so the no-wraparound guarantee holds on every supported GOARCH. The change is package-internal; IncomingFrame has no callers outside pkg/execution.
jason810496
added a commit
to jason810496/airflow
that referenced
this pull request
May 29, 2026
…dings - Omit map_index from SetXCom for unmapped task instances and carry it as *int from api.TaskInstance through to SetXComMsg. Fixes a compile-time mismatch and matches the supervisor's "absent vs -1" semantics for unmapped tasks. - Honor context cancellation in CoordinatorComm.Communicate by running the request send in a goroutine and selecting on ctx.Done(), so a blocked supervisor socket no longer wedges the caller. The underlying connection is left alone on cancel to avoid poisoning future writes with a stale deadline or leaving a partial length-prefixed frame on the wire. - Decode ConnectionResult.Login and Password as *string via a new mapStringPtr helper so an explicitly empty credential round-trips through the coordinator client instead of being silently treated as absent. sdk.Connection already encodes the distinction. Adds regression tests for each case. go-sdk: drop unused details field from CoordinatorClient CoordinatorClient stored a *StartupDetails it never read, and the companion PRs (apache#67155, apache#67318) that also construct or pass details into NewCoordinatorClient never read c.details either — they work directly off the local *StartupDetails inside RunTask. Keeping the field as API surface implies a contract the type does not honor, so remove it and let callers pass only the comm channel. go-sdk: unify ErrorResponse decoding in CoordinatorComm.Communicate The dispatcher response could carry an error in two places — the third element of a 3-tuple response frame, or as the body of a 2-tuple frame whose "type" is "ErrorResponse" — and Communicate inspected each path independently. The two branches diverged on nil-guarding decodeErrorResponse, which was easy-to-miss latent inconsistency: either path could grow a bug the other did not. Extract the source selection into errMapFromFrame so the decode and *ApiError construction live in one place. go-sdk: widen frame IDs to int64 end-to-end CoordinatorComm's request-id counter was atomic.Int64 (chosen so a long-running runtime cannot wrap), but the value was narrowed to int when stored in the pending map and IncomingFrame.ID. On 32-bit GOARCH that narrowing reintroduces the wraparound the int64 counter was meant to prevent, and the comment promising "wide enough to avoid wraparound" becomes architecture-dependent. Widen IncomingFrame.ID, the pending map key, encodeRequest, and the test fixtures to int64 so the no-wraparound guarantee holds on every supported GOARCH. The change is package-internal; IncomingFrame has no callers outside pkg/execution.
jason810496
added a commit
that referenced
this pull request
May 30, 2026
#67317) * go-sdk: Add concurrent-safe coordinator comms, log handler, and client Build the comm layer on top of the protocol primitives so subsequent runtime code has a single typed entry point for talking to the supervisor. CoordinatorComm runs a concurrent-safe dispatcher loop that fans inbound frames out to per-request reply channels keyed by a monotonic id, propagates context cancellation, and cleans up pending requests on SendRequest failure. SocketLogHandler streams slog records as structured JSON over the dedicated logs socket so the supervisor can demux task logs without parsing stderr. CoordinatorClient implements the sdk.Client surface (GetVariable honouring AIRFLOW_VAR_* overrides, GetConnection, XCom push/pull, deferral) by routing each method through the dispatcher and translating supervisor not-found responses into the SDK's sentinel errors. No server or task-runner loop is wired yet -- that lands in the next PR in this stack. * self-review: Address coordinator comms, client, and logger review findings - Omit map_index from SetXCom for unmapped task instances and carry it as *int from api.TaskInstance through to SetXComMsg. Fixes a compile-time mismatch and matches the supervisor's "absent vs -1" semantics for unmapped tasks. - Honor context cancellation in CoordinatorComm.Communicate by running the request send in a goroutine and selecting on ctx.Done(), so a blocked supervisor socket no longer wedges the caller. The underlying connection is left alone on cancel to avoid poisoning future writes with a stale deadline or leaving a partial length-prefixed frame on the wire. - Decode ConnectionResult.Login and Password as *string via a new mapStringPtr helper so an explicitly empty credential round-trips through the coordinator client instead of being silently treated as absent. sdk.Connection already encodes the distinction. Adds regression tests for each case. go-sdk: drop unused details field from CoordinatorClient CoordinatorClient stored a *StartupDetails it never read, and the companion PRs (#67155, #67318) that also construct or pass details into NewCoordinatorClient never read c.details either — they work directly off the local *StartupDetails inside RunTask. Keeping the field as API surface implies a contract the type does not honor, so remove it and let callers pass only the comm channel. go-sdk: unify ErrorResponse decoding in CoordinatorComm.Communicate The dispatcher response could carry an error in two places — the third element of a 3-tuple response frame, or as the body of a 2-tuple frame whose "type" is "ErrorResponse" — and Communicate inspected each path independently. The two branches diverged on nil-guarding decodeErrorResponse, which was easy-to-miss latent inconsistency: either path could grow a bug the other did not. Extract the source selection into errMapFromFrame so the decode and *ApiError construction live in one place. go-sdk: widen frame IDs to int64 end-to-end CoordinatorComm's request-id counter was atomic.Int64 (chosen so a long-running runtime cannot wrap), but the value was narrowed to int when stored in the pending map and IncomingFrame.ID. On 32-bit GOARCH that narrowing reintroduces the wraparound the int64 counter was meant to prevent, and the comment promising "wide enough to avoid wraparound" becomes architecture-dependent. Widen IncomingFrame.ID, the pending map key, encodeRequest, and the test fixtures to int64 so the no-wraparound guarantee holds on every supported GOARCH. The change is package-internal; IncomingFrame has no callers outside pkg/execution. * go-sdk: Fix SocketLogHandler WithAttrs/WithGroup ordering Pre-group attrs were silently re-qualified by any later WithGroup call, violating slog.Handler's "subsequently-added attrs only" contract. Snapshot the group prefix when WithAttrs is called and freeze each attr's key, so later WithGroup affects only record-level attrs and attrs added afterwards. Also document the deliberate flat dotted-key wire format on SocketLogHandler so it isn't switched to nested JSON objects without coordinating with the supervisor-side log parser. * fixup: Expand inline slog groups and lower late-reply log level SocketLogHandler dropped inline slog.Group attributes: a KindGroup value resolved to []slog.Attr that marshaled to "{}", so task code logging with slog.Group lost data. Handle now recurses into group values, expanding them into dotted keys (req.method) the same way WithGroup does, and skips empty groups per the slog.Handler contract. Also lower the "Discarding frame with no matching waiter" log from Warn to Debug: Communicate deliberately drops a waiter on cancel/timeout, so a late reply landing here is the expected deadline path, not a protocol bug.
1 task
jason810496
force-pushed
the
refactor/go-sdk/dag-authoring-api
branch
from
June 3, 2026 03:14
c3a2584 to
fd01c17
Compare
4 tasks
- Add `messages_test.go` to test message decoding and encoding functionalities. - Introduce `serde.go` for serialization of various data types to Airflow's format. - Create `serde_test.go` to validate serialization logic and ensure correctness. - Implement `server.go` to handle communication with the supervisor and manage task execution. - Add `task_runner.go` to execute tasks based on received startup details and handle success/failure.
…proved configuration flexibility
Re-add the DagFileParseRequest message, its decoder wiring in decodeIncomingBody, and the server dispatch case so the supervisor can request DAG parsing and receive a DagFileParsingResult. Also fix two AddTaskWithName test calls broken by the registration signature change.
Align the Go bundle serializer with Python's serialize_dag so a Go-authored Dag produces the same serialized shape Python would for the fields the SDK models, verified field-by-field against a reference dump: - Always emit the DAG fields that have no JSON-schema default and that Python therefore never omits: catchup, disable_bundle_versioning, max_active_tasks, max_active_runs (default 16 from [core]) and max_consecutive_failed_dag_runs. - Sort tags, matching Python's set-backed serialization (stable dag_hash). - Always emit template_fields on each task. Cron schedules still serialize to CronTriggerTimetable, which matches only the default [scheduler] create_cron_data_intervals=False; honoring the non-default value needs the supervisor to pass scheduler config over the coordinator protocol, tracked in apache#67938.
Replace the hand-written TaskSpec struct and its hand-mapped serializer omit-if-default rules with spec.gen.go, generated from the "operator" definition in airflow-core/src/airflow/serialization/schema.json by a local gen tool (same pattern as pkg/execution/genmodels). Field types and schema defaults can no longer drift from what the scheduler deserializes; regenerate with `just generate-specs`. DagSpec and the registration Info structs move to the hand-written spec.go next to it: Schedule is an SDK-level concept the schema has no scalar for, and several dag keys are always-emitted with [core]-config fallbacks the schema cannot express.
jason810496
force-pushed
the
refactor/go-sdk/dag-authoring-api
branch
from
July 21, 2026 03:17
1a47ba8 to
8b8f7f3
Compare
The generator read field types and defaults from schema.json, but the field list itself was a hand-written allowlist, so a new author-settable operator key would go silently missing from TaskSpec. Selecting every eligible scalar key from the schema - with serializer-owned and Python-only keys excluded by documented rules - inverts the drift direction: new schema keys surface in the regenerated spec.gen.go diff, and stale exclusions fail generation. This also surfaces ignore_first_depends_on_past and wait_for_past_depends_before_skipping, which the hand list had missed.
The hand-written DagSpec duplicated the "dag" definition of the serialization schema, and its emit policy lived apart from it in the serde layer, so both could silently drift from what the scheduler deserializes. Deriving the struct and its SchemaFields policy from the schema keeps the exposed dag fields honest the same way TaskSpec already is; the rules the schema cannot express — the always-emit keys with their [core]-config fallbacks, the nullable is_paused_upon_creation, sorted tags, and the SDK-only Schedule field — are declared per key in the generator config and fail generation when they go stale.
1 task
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
DagImporterinterface)Why
While waiting for the AIP-85
DagImporterinterface, we can already build the "define native Dags in a Lang-SDK" side in the Go SDK, so the native Dag feature can ship as soon as the Core / Task-SDK side is ready. This work is fully independent of and parallel to theDagImportereffort.Example
Here's what exactly the syntax will look like before supporting Dag-level TaskFlow.
How
DagSpecandTaskSpecare generated fromairflow-core/src/airflow/serialization/schema.json(thedag/operatordefinitions), so the airflow-core serialization schema stays the single source of truth for the Go SDK — other Lang-SDKs will follow the same pattern. The few rules the schema cannot express (always-emit keys with[core]config fallbacks, nullable booleans, sorted tags, the SDK-onlySchedule) live in small generator tables that fail generation when they go stale.go-sdk/pkg/execution/serde.goserializes registered Dags to the same v3 output Python'sDagSerializationproduces, andgo-sdk/pkg/execution/dag_parser.goanswers the coordinator'sDagFileParseRequestwith aDagFileParsingResultbuilt from it.DagImporterinterface is wired up — no breaking change.What
bundle/bundlev1/registry.go):AddDagtakes an optionalDagSpec(schedule, tags, timeouts, …);AddTask/AddTaskWithNametake an optionalTaskSpec(queue, retries, …) plus a downstream task-id list, so multi-task Dags can be authored in pure Go.bundle/bundlev1/gen,spec.gen.go): both spec structs and theirSchemaFields()emit policies derive fromschema.json; regenerate withjust generate-specs.pkg/execution/serde.go): emits DagSerialization v3 matching Python field-for-field (timetable fromSchedule, omit-if-default, always-emit keys, sorted tags, epoch-encoded dates).pkg/execution/server.go,dag_parser.go): the first comm-socket frame selects one-shot Dag parsing (DagFileParseRequest) vs task execution (StartupDetails).Was generative AI tooling used to co-author this PR?