pool: Enhance worker lifecycle management and job reliability #51
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Ensure two nodes can't process stale workers concurrently
Instead of each worker having their own cache, have the parent node hold the cache to avoid duplicate caching.
The original code used future timestamps in workerKeepAliveMap to prevent concurrent cleanup operations. This made stale workers appear active and could permanently prevent cleanup if a node crashed during the process. Fixed by: - Added dedicated cleanupMap to track workers being cleaned up - Implemented proper concurrency handling using SetIfNotExists/TestAndSet - Added retry logic with exponential backoff for requeuing jobs - Ensured cleanup map is properly closed during node shutdown - Updated worker.go to handle new processRequeuedJobs retry parameter The fix ensures stale workers and their jobs are reliably cleaned up even in case of node failures or concurrent cleanup attempts.
- Use background context for worker goroutines to prevent premature termination - Preserve logging context while making worker lifecycle independent of caller - Rename maps for better clarity (e.g. jobsMap -> jobMap) - Improve node stream management with nodeStreams map - Clean up error handling and logging patterns This fixes an issue where workers could be leaked when the caller's context was cancelled before proper cleanup could occur.
Improves the Node component's ability to detect and reassign jobs from stale or deleted workers by: 1. Adding explicit orphaned job detection for workers missing keep-alive entries 2. Centralizing worker cleanup logic to ensure consistent job reassignment 3. Simplifying worker state validation to catch edge cases in distributed scenarios This ensures that no jobs are lost when workers become unavailable, maintaining eventual consistency of job assignments across the worker pool.
Enhances worker cleanup mechanism by handling stale cleanup locks and adding cleanup verification. Key changes: * Add detection and cleanup of stale worker cleanup locks * Clean up jobs from jobMap after successful requeue * Improve logging around worker cleanup and job requeuing * Upgrade requeue log level to Info for better operational visibility This improves reliability of the distributed job system by preventing orphaned jobs and stale locks from accumulating over time.
Add cleanup of stale consumers during sink initialization to prevent accumulation of stale consumers in Redis. Previously stale consumers were only cleaned up periodically, which could lead to a buildup if sinks did not shut down cleanly. Also refactor the stale consumer cleanup logic to: 1. Extract common cleanup code into deleteStreamStaleConsumers 2. Improve error handling and logging 3. Properly clean up all related data structures (Redis consumer group, keep-alive map, and consumers map)
Improve the worker cleanup implementation by: 1. Split cleanupWorker into smaller, focused functions: - acquireCleanupLock: handles cleanup lock management - requeueWorkerJobs: handles job requeuing - cleanupWorker: orchestrates the cleanup process 2. Simplify cleanupInactiveWorkers: - Use activeWorkers() to get list of active workers - Combine jobMap and workerMap checks into a single loop - Skip workers being actively cleaned up 3. Rename isActive to isWithinTTL to better reflect its purpose - Function checks if a timestamp is within TTL duration - Used consistently across node and worker cleanup
This commit adds a new background process to clean up stale entries in the pending jobs map. Previously, stale entries were only cleaned up when attempting to dispatch a job with the same key. Now, a dedicated goroutine runs at the ackGracePeriod frequency to proactively remove expired entries. Additional changes: - Fix jobPendingMap comment to clarify it's indexed by job key not worker ID - Add debug logs for worker shutdown in handleEvents and keepAlive - Refactor timestamp validation to use isWithinTTL helper across the codebase - Improve error handling in cleanupStalePendingJobs using TestAndDelete The periodic cleanup helps prevent memory leaks from abandoned dispatch attempts and makes the job dispatch system more reliable.
Remove the requeueJob helper function and directly use dispatchJob for requeueing jobs during worker cleanup and rebalancing.
Non-read notifications on the ichan channel were blocking writes and causing deadlocks during Set operations. This commit removes ichan and replaces it with a waiter-based mechanism using dedicated per-call channels, ensuring notifications are delivered without blocking.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR solves a potential deadlock with maps and improves the stability and reliability of the worker pool by making several key improvements.
Worker & Job Cleanup:
Concurrency & Race Conditions:
Architecture: