fix: listener v2 status race #1856
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
There was a problem hiding this comment.
Pull Request Overview
This PR introduces a background polling goroutine in ListenV2 to keep a worker’s active status true whenever any connection remains open, preventing unintended deactivation due to timestamp races.
- Starts a ticker-based goroutine that checks and reactivates a worker every 2 seconds
- Logs debug details on poller shutdown and status updates
- Handles database errors gracefully during polling
Comments suppressed due to low confidence (1)
internal/services/dispatcher/server.go:507
- Add unit or integration tests for this new polling behavior, verifying that the worker remains active while any listener is connected and that status updates use the correct timestamp ordering.
go func() {
| // If worker is not active, set it to active | ||
| if !currentWorker.IsActive { | ||
| s.l.Debug().Msgf("worker %s active status poller: setting worker to active", request.WorkerId) | ||
| _, err := s.repo.Worker().UpdateWorkerActiveStatus(ctx, tenantId, request.WorkerId, true, sessionEstablished) |
There was a problem hiding this comment.
The polling goroutine reuses the original sessionEstablished timestamp, which can be older than later updates and reintroduce the race condition. Use the current time (e.g., time.Now()) for each update to ensure the timestamp always increases.
| _, err := s.repo.Worker().UpdateWorkerActiveStatus(ctx, tenantId, request.WorkerId, true, sessionEstablished) | |
| _, err := s.repo.Worker().UpdateWorkerActiveStatus(ctx, tenantId, request.WorkerId, true, time.Now()) |
| // This polling goroutine ensures that as long as ANY ListenV2 connection is active, | ||
| // the worker status remains active=true, preventing the race condition | ||
| go func() { | ||
| ticker := time.NewTicker(2 * time.Second) // 2x heartbeat interval |
There was a problem hiding this comment.
[nitpick] Avoid hardcoding the polling interval. Extract 2 * time.Second into a named constant or configuration to ensure it stays in sync with your heartbeat settings.
| ticker := time.NewTicker(2 * time.Second) // 2x heartbeat interval | |
| ticker := time.NewTicker(pollingInterval) // 2x heartbeat interval |
| s.workers.DeleteForSession(request.WorkerId, sessionId) | ||
| }() | ||
|
|
||
| // HOTFIX: Start a background goroutine to poll and ensure worker stays active |
There was a problem hiding this comment.
[nitpick] This large inline comment could be moved to a dedicated helper or summarized more concisely to improve readability. Consider extracting the poller into its own function with a clear docstring.
| // This addresses the race condition where multiple listeners can interfere with each other's active status | ||
| // | ||
| // Race condition scenario: | ||
| // 1. Worker A establishes ListenV2 connection, sets worker active=true with timestamp T1 |
There was a problem hiding this comment.
I would update this comment to indicate that it is the same worker that's establishing the connection, this is incorrect as written
Description
This addresses the race condition where multiple listeners can interfere with each other's active status
Race condition scenario:
This polling goroutine ensures that as long as ANY ListenV2 connection is active,
the worker status remains active=true, preventing the race condition
Type of change