mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Fix async poller helpers in versioning_3_test.go (#9643)
## What changed?
Fix four async poller helpers in `versioning_3_test.go` to avoid calling
`s.NoError()` inside goroutines that can outlive the test:
- `doPollWftAndHandle `
- `pollWftAndHandleQueries`
- `pollNexusTaskAndHandle`
- `doPollActivityAndHandle`
Each helper had this pattern: a closure `f()` that called
`s.NoError(err)`, then in the async path the closure was invoked from a
goroutine. Moving `s.NoError `out of the closure into the sync-only path
eliminates the assertion-in-goroutine risk.
## Why?
These goroutines poll with up to 30–60 second timeouts, but the
enclosing tests use 15-second contexts. When the test context expires
and the test completes first, the goroutine is still running. When its
poll eventually times out and it calls s.NoError(err), Go panics:
`panic: Fail in goroutine after
TestVersioning3FunctionalSuiteV2/TestWorkflowWithPinnedOverride_NoSticky
has completed`
This crash kills the entire test binary, causing `gotestsum` to write
only `junit.crash.xml` with no individual test results. Observed in CI
run #23502599693 on multiple backends (sqlite shard2, postgres12_pgx
shard0, cass_os3 shard0).
## How did you test it?
- [ ] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)
## Potential risks
None, test only change.
This commit is contained in:
+22
-22
@@ -3877,25 +3877,25 @@ func (s *Versioning3Suite) doPollWftAndHandle(
|
||||
handler func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error),
|
||||
) (*taskpoller.TaskPoller, *workflowservice.RespondWorkflowTaskCompletedResponse) {
|
||||
poller := taskpoller.New(s.T(), s.FrontendClient(), s.Namespace().String())
|
||||
f := func() *workflowservice.RespondWorkflowTaskCompletedResponse {
|
||||
f := func() (*workflowservice.RespondWorkflowTaskCompletedResponse, error) {
|
||||
tq := tv.TaskQueue()
|
||||
if sticky {
|
||||
tq = tv.StickyTaskQueue()
|
||||
}
|
||||
resp, err := poller.PollWorkflowTask(
|
||||
return poller.PollWorkflowTask(
|
||||
&workflowservice.PollWorkflowTaskQueueRequest{
|
||||
DeploymentOptions: tv.WorkerDeploymentOptions(versioned),
|
||||
TaskQueue: tq,
|
||||
},
|
||||
).HandleTask(tv, handler, taskpoller.WithTimeout(30*time.Second))
|
||||
s.NoError(err)
|
||||
return resp
|
||||
}
|
||||
if async == nil {
|
||||
return poller, f()
|
||||
resp, err := f()
|
||||
s.NoError(err)
|
||||
return poller, resp
|
||||
} else {
|
||||
go func() {
|
||||
f()
|
||||
_, _ = f() // errors are surfaced via test context timeout on WaitForChannel
|
||||
close(async)
|
||||
}()
|
||||
}
|
||||
@@ -3909,25 +3909,25 @@ func (s *Versioning3Suite) pollWftAndHandleQueries(
|
||||
handler func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondQueryTaskCompletedRequest, error),
|
||||
) (*taskpoller.TaskPoller, *workflowservice.RespondQueryTaskCompletedResponse) {
|
||||
poller := taskpoller.New(s.T(), s.FrontendClient(), s.Namespace().String())
|
||||
f := func() *workflowservice.RespondQueryTaskCompletedResponse {
|
||||
f := func() (*workflowservice.RespondQueryTaskCompletedResponse, error) {
|
||||
tq := tv.TaskQueue()
|
||||
if sticky {
|
||||
tq = tv.StickyTaskQueue()
|
||||
}
|
||||
resp, err := poller.PollWorkflowTask(
|
||||
return poller.PollWorkflowTask(
|
||||
&workflowservice.PollWorkflowTaskQueueRequest{
|
||||
DeploymentOptions: tv.WorkerDeploymentOptions(true),
|
||||
TaskQueue: tq,
|
||||
},
|
||||
).HandleLegacyQuery(tv, handler)
|
||||
s.NoError(err)
|
||||
return resp
|
||||
}
|
||||
if async == nil {
|
||||
return poller, f()
|
||||
resp, err := f()
|
||||
s.NoError(err)
|
||||
return poller, resp
|
||||
}
|
||||
go func() {
|
||||
f()
|
||||
_, _ = f() // errors are surfaced via test context timeout on WaitForChannel
|
||||
close(async)
|
||||
}()
|
||||
return nil, nil
|
||||
@@ -3940,25 +3940,25 @@ func (s *Versioning3Suite) pollNexusTaskAndHandle(
|
||||
handler func(task *workflowservice.PollNexusTaskQueueResponse) (*workflowservice.RespondNexusTaskCompletedRequest, error),
|
||||
) (*taskpoller.TaskPoller, *workflowservice.RespondNexusTaskCompletedResponse) {
|
||||
poller := taskpoller.New(s.T(), s.FrontendClient(), s.Namespace().String())
|
||||
f := func() *workflowservice.RespondNexusTaskCompletedResponse {
|
||||
f := func() (*workflowservice.RespondNexusTaskCompletedResponse, error) {
|
||||
tq := tv.TaskQueue()
|
||||
if sticky {
|
||||
tq = tv.StickyTaskQueue()
|
||||
}
|
||||
resp, err := poller.PollNexusTask(
|
||||
return poller.PollNexusTask(
|
||||
&workflowservice.PollNexusTaskQueueRequest{
|
||||
DeploymentOptions: tv.WorkerDeploymentOptions(true),
|
||||
TaskQueue: tq,
|
||||
},
|
||||
).HandleTask(tv, handler, taskpoller.WithTimeout(10*time.Second))
|
||||
s.NoError(err)
|
||||
return resp
|
||||
}
|
||||
if async == nil {
|
||||
return poller, f()
|
||||
resp, err := f()
|
||||
s.NoError(err)
|
||||
return poller, resp
|
||||
}
|
||||
go func() {
|
||||
f()
|
||||
_, _ = f() // errors are surfaced via test context timeout on WaitForChannel
|
||||
close(async)
|
||||
}()
|
||||
return nil, nil
|
||||
@@ -3987,19 +3987,19 @@ func (s *Versioning3Suite) doPollActivityAndHandle(
|
||||
handler func(task *workflowservice.PollActivityTaskQueueResponse) (*workflowservice.RespondActivityTaskCompletedRequest, error),
|
||||
) {
|
||||
poller := taskpoller.New(s.T(), s.FrontendClient(), s.Namespace().String())
|
||||
f := func() {
|
||||
f := func() error {
|
||||
_, err := poller.PollActivityTask(
|
||||
&workflowservice.PollActivityTaskQueueRequest{
|
||||
DeploymentOptions: tv.WorkerDeploymentOptions(versioned),
|
||||
},
|
||||
).HandleTask(tv, handler, taskpoller.WithTimeout(time.Minute))
|
||||
s.NoError(err)
|
||||
return err
|
||||
}
|
||||
if async == nil {
|
||||
f()
|
||||
s.NoError(f())
|
||||
} else {
|
||||
go func() {
|
||||
f()
|
||||
_ = f() // errors are surfaced via test context timeout on WaitForChannel
|
||||
close(async)
|
||||
}()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user