Repository navigation
Reject invalid Go SDK task handler registrations up front - #73549
Conversation
There was a problem hiding this comment.
https://github.com/orgs/apache/projects/499/views/7?pane=issue&itemId=248986989
- Registering after
Servehas been called panics - registration closes when serving starts.- A duplicate
(dagId, taskId)panics.- A struct carrying
arg:tags mixed with other data parameters is rejected (an untagged struct has no such guard and is decoded positionally as one value).- Signature errors - non-function, wrong return shape, missing leading
airflow.Context- panic with the offending function named.
It seems the scope of current PR is not align with the issue scope.
How about checking whether all the above issue scope are covered instead of having current refactor.
Additionally, the task id validation should be in compile time instead of at the task runtime. IIUC, the current refactor will introduce side-effect (might raise validation error) during the runtime.
A variadic function passed the signature check as if its ... parameter were an ordinary slice, so it registered cleanly and then panicked inside reflect the first time the supervisor ran it: Execute calls the function with Call, which hands a variadic parameter the slice itself instead of spreading it. There is no argument shape the runtime can bind to a ... parameter, so the registration cannot become valid later.
ADR 0007 states that registration closes when Serve is called, but nothing enforced it: a Register left below Serve in main silently added a handler to the map the runtime was already answering from, so what the bundle could run depended on how far main had got. Serve closes registration whatever the run goes on to do, since the mistake is where Register sits in main rather than what that run turned out to be.
9a69eae to
2e058b5
Compare
Thanks I had misunderstood earlier. |
jason810496
left a comment
jason810496
left a comment
There was a problem hiding this comment.
Nice, thanks for the update.
The flag describes whether the bundle still takes registrations, not the state of one of the maps it holds. Keeping it on the handler map meant every further kind of registration would need a flag of its own and a matching line in Serve, so a native Dag would have arrived with the check missing until someone remembered it. The check now runs before Register looks at what the item is, so it covers a kind added later for free. It reads an atomic rather than taking a lock because the handler map keeps its own — the runtime is handed that map on its own — and a second mutex would introduce a lock order to respect.
jason810496
left a comment
jason810496
left a comment
There was a problem hiding this comment.
Thanks for the update. LGTM now.
2 participants
Toward "make every registration-time mistake fail at registration rather than on first execution". Checked against the scope,
mainalready satisfies two of its four points, so this PR closes the gap in the other two.Why
Registration did not close at
Serve. ADR 0007 states that it does and that a laterRegisteris a programming error that panics, but nothing enforced it.BundleRefcarried no such state, so aRegisterleft belowServeinmainsilently added a handler to the map the runtime was already answering from — what the bundle could run depended on how farmainhad got.A variadic task function registered cleanly and then panicked on first execution.
func(actx airflow.Context, names ...string) errorpassed the signature check as if its...parameter were an ordinary[]string.Executecalls the function withCall, which hands a variadic parameter the slice itself instead of spreading it, so the first run under the supervisor panicked inside reflect. No argument shape can fill a...parameter on this path, so the registration can never become valid;[]stringtakes the same values and works.What
go-sdk/airflow/bundle.go—taskHandlerMapgains aclosedfield and aclosemethod. The check sits insideadd, under the mutex that already guards the map, so there is no window between reading the flag and writing the handler.go-sdk/airflow/serve.go—servecloses registration on entry, before parsing flags, so it also applies to a run that only prints usage or fails on a flag. The mistake is whereRegistersits inmain, not what that run turned out to do.go-sdk/internal/bundle/task.go—validateFnrejects a variadicfn, placed after theKindcheck thatreflect.Type.IsVariadicrequires.RegisterandServenow state that registration closes.TestRegisterAfterServePanics,TestRegisterAfterAFailedServePanics, and avariadiccase in the existingTestTaskHandlerPanicsOnBadHandlertable.go test ./... -count=1andgo test ./airflow/ -racepass; the concurrent-registration test still passes with the new field.Was generative AI tooling used to co-author this PR?