mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
Kannan RajahandGitHub
b24e499178
## What changed? Eagerly remove worker from pollerHistory during graceful shutdown so DescribeTaskQueue doesn't show stale pollers. **Note:** This PR subtly changes how pollerHistory is updated `before` and `after` a poll request. Before: - UpdatePollerInfo called at poll START - defer UpdatePollerInfo called at poll END (always, regardless of how poll ended) After: - UpdatePollerInfo called at poll START - UpdatePollerInfo called at poll END only if ctx.Err() != context.Canceled Skipped on cancellation (shutdown/disconnect) to avoid re-adding entry after RemovePoller The defer was originally added (PR #2811) to keep timestamps fresh, but HasPollerAfter() first checks currentPolls > 0 for active polls, so the timestamp is a secondary check **Main changes** API - request_response.proto: Added worker_identity field to CancelOutstandingWorkerPollsRequest Implementation - matching_engine.go: As part of CancelOutstandingWorkerPolls, also invoke RemovePoller on the task queue partition manager. - task_queue_partition_manager.go: Forward the removal to default and versioned task queues. ## Why? Previously, pollers lingered in pollerHistory until TTL expired (~5 min), showing workers that already shut down. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) ## Potential risks Controlled by dynamic config: EnableCancelWorkerPollsOnShutdown
62 lines
1.5 KiB
Go
62 lines
1.5 KiB
Go
package matching
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
func TestPollerHistory(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
t.Run("removePoller removes entry", func(t *testing.T) {
|
|
t.Parallel()
|
|
history := newPollerHistory(5 * time.Minute)
|
|
|
|
identity := pollerIdentity("worker-1@host")
|
|
history.updatePollerInfo(identity, &pollMetadata{})
|
|
|
|
// Verify it exists
|
|
pollers := history.getPollerInfo(time.Time{})
|
|
require.Len(t, pollers, 1)
|
|
require.Equal(t, "worker-1@host", pollers[0].Identity)
|
|
|
|
// Remove it
|
|
history.removePoller(identity)
|
|
|
|
// Verify it's gone
|
|
pollers = history.getPollerInfo(time.Time{})
|
|
require.Empty(t, pollers)
|
|
})
|
|
|
|
t.Run("removePoller is idempotent", func(t *testing.T) {
|
|
t.Parallel()
|
|
history := newPollerHistory(5 * time.Minute)
|
|
|
|
identity := pollerIdentity("worker-1@host")
|
|
history.updatePollerInfo(identity, &pollMetadata{})
|
|
|
|
// Remove twice - should not panic
|
|
history.removePoller(identity)
|
|
history.removePoller(identity)
|
|
|
|
pollers := history.getPollerInfo(time.Time{})
|
|
require.Empty(t, pollers)
|
|
})
|
|
|
|
t.Run("removePoller does not affect other entries", func(t *testing.T) {
|
|
t.Parallel()
|
|
history := newPollerHistory(5 * time.Minute)
|
|
|
|
history.updatePollerInfo(pollerIdentity("worker-1@host"), &pollMetadata{})
|
|
history.updatePollerInfo(pollerIdentity("worker-2@host"), &pollMetadata{})
|
|
|
|
history.removePoller(pollerIdentity("worker-1@host"))
|
|
|
|
pollers := history.getPollerInfo(time.Time{})
|
|
require.Len(t, pollers, 1)
|
|
require.Equal(t, "worker-2@host", pollers[0].Identity)
|
|
})
|
|
}
|