Go-SDK: Add concurrent-safe coordinator comms, log handler, and client (#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. J
Jason(Zhe-You) Liu committed
70d269aadb7a7f828ad58f1f9ae7b2921abb6071
Parent: 62d60e4
Committed by GitHub <noreply@github.com>
on 5/30/2026, 10:52:57 AM