hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix data race in workerdeployment unit tests (#9766)

Browse Source 
## What changed?
Replace plain `int` with `atomic.Int32` in two VersionWorkflowSuite unit
tests.

## Why?
CI caught this as a crash in the unit test job

Failure
[here](https://github.com/temporalio/temporal/actions/runs/23856744785)
```
 WARNING: DATA RACE                                                                                                                                                                                      
  Read at 0x00c000b005e8 by goroutine 320:                                    
    go.temporal.io/server/service/worker/workerdeployment.(*VersionWorkflowSuite).Test_MultipleSyncStates_BlocksCaNUntilAllComplete.func2()                                                               
        version_workflow_test.go:1240 +0x2c                                                                                                                                                               
  Previous write at 0x00c000b005e8 by goroutine 326:                                                                                                                                                      
    ...version_workflow_test.go:1240 +0x3c  
```

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [X] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
NA
This commit is contained in:
Sean Kane
2026-04-01 15:30:55 -04:00
committed by GitHub
parent a9316e8f6b
commit e379e36cea
1 changed files with 13 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/worker/workerdeployment/version_workflow_test.go
+13 -12
View File
@@ -3,6 +3,7 @@ package workerdeployment
import (
"context"
"fmt"
"sync/atomic"
"testing"
"time"
@@ -1223,21 +1224,21 @@ func (s *VersionWorkflowSuite) Test_MultipleSyncStates_BlocksCaNUntilAllComplete
taskQueueName := tv.TaskQueue().Name
syncCallCount := 0
var syncCallCount atomic.Int32
s.env.OnActivity(a.SyncDeploymentVersionUserData, mock.Anything, mock.Anything).Return(
func(ctx context.Context, req *deploymentspb.SyncDeploymentVersionUserDataRequest) (*deploymentspb.SyncDeploymentVersionUserDataResponse, error) {
syncCallCount++
count := syncCallCount.Add(1)
return &deploymentspb.SyncDeploymentVersionUserDataResponse{
TaskQueueMaxVersions: map[string]int64{taskQueueName: int64(syncCallCount * 10)},
TaskQueueMaxVersions: map[string]int64{taskQueueName: int64(count * 10)},
}, nil
},
).Maybe()
// Mock propagation check with delay to simulate slow propagation
propagationCheckCount := 0
var propagationCheckCount atomic.Int32
s.env.OnActivity(a.CheckWorkerDeploymentUserDataPropagation, mock.Anything, mock.Anything).
Return(func(ctx context.Context, req *deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest) error {
propagationCheckCount++
propagationCheckCount.Add(1)
return nil
}).
After(100 * time.Millisecond).
@@ -1319,7 +1320,7 @@ func (s *VersionWorkflowSuite) Test_MultipleSyncStates_BlocksCaNUntilAllComplete
s.True(s.env.IsWorkflowCompleted())
// Both propagations should have completed before CaN
s.Equal(2, propagationCheckCount, "Both propagations should complete before CaN")
s.Equal(2, int(propagationCheckCount.Load()), "Both propagations should complete before CaN")
}
// Test_SyncState_And_RegisterWorker_ConcurrentPropagations tests concurrent async propagations
@@ -1335,10 +1336,10 @@ func (s *VersionWorkflowSuite) Test_SyncState_And_RegisterWorker_ConcurrentPropa
taskQueueName := tv.TaskQueue().Name
newTaskQueueName := tv.TaskQueue().Name + "_new"
syncActivityCalls := 0
var syncActivityCalls atomic.Int32
s.env.OnActivity(a.SyncDeploymentVersionUserData, mock.Anything, mock.Anything).Return(
func(ctx context.Context, req *deploymentspb.SyncDeploymentVersionUserDataRequest) (*deploymentspb.SyncDeploymentVersionUserDataResponse, error) {
syncActivityCalls++
syncActivityCalls.Add(1)
if req.UpdateRoutingConfig != nil && len(req.Sync) == 1 && req.Sync[0].Name == taskQueueName {
// This is the SyncState propagation
return &deploymentspb.SyncDeploymentVersionUserDataResponse{
@@ -1352,10 +1353,10 @@ func (s *VersionWorkflowSuite) Test_SyncState_And_RegisterWorker_ConcurrentPropa
},
).Maybe()
propagationChecks := 0
var propagationChecks atomic.Int32
s.env.OnActivity(a.CheckWorkerDeploymentUserDataPropagation, mock.Anything, mock.Anything).
Return(func(ctx context.Context, req *deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest) error {
propagationChecks++
propagationChecks.Add(1)
return nil
}).
After(50 * time.Millisecond).
@@ -1438,9 +1439,9 @@ func (s *VersionWorkflowSuite) Test_SyncState_And_RegisterWorker_ConcurrentPropa
s.True(s.env.IsWorkflowCompleted())
// Both syncs should complete
s.GreaterOrEqual(syncActivityCalls, 2, "Both propagations should complete before CaN")
s.GreaterOrEqual(int(syncActivityCalls.Load()), 2, "Both propagations should complete before CaN")
// Only syncVersionState should wait for propagation
s.GreaterOrEqual(propagationChecks, 1, "Both propagations should complete before CaN")
s.GreaterOrEqual(int(propagationChecks.Load()), 1, "Both propagations should complete before CaN")
}
// Test_SyncState_SignalsPropagationComplete_WithCorrectRevisionNumber tests that the deployment