feat: implement housekeeper election and leadership - #76
Open
lucasoares wants to merge 18 commits into
Open
Conversation
Co-authored-by: lucasoares <10624972+lucasoares@users.noreply.github.com>
Co-authored-by: lucasoares <10624972+lucasoares@users.noreply.github.com>
scheduleTask and unlockMessages were left over from before this PR introduced scheduleTaskWithDistributedLock and unlockMessagesParallel and were no longer called anywhere.
ReleaseLock and RefreshLock used a non-atomic Get-then-Del/Expire sequence to check lock ownership. Between the Get and the mutating call, the lock could expire and be re-acquired by another instance, causing this instance to release or refresh a lock it no longer owns. Add CompareAndDelete/CompareAndExpire to the Cache interface, backed by atomic Lua scripts in Redis (GET+DEL / GET+PEXPIRE) and a lock-guarded compare in MemoryCache, and use them in RedisDistributedLock.ReleaseLock/RefreshLock so ownership is checked and mutated atomically.
# Conflicts: # internal/queue/cache/redis_cache.go
There was a problem hiding this comment.
Pull request overview
This PR adds distributed coordination for housekeeper tasks (via Redis-backed locks) and introduces bounded parallelism for unlocking, aiming to support multiple housekeeper instances without conflicting executions.
Changes:
- Added a
DistributedLockabstraction with a Redis implementation using atomic compare-and-delete / compare-and-expire primitives. - Updated housekeeper scheduling to use distributed locks (and added metrics “leader” gating).
- Parallelized unlock processing with configurable concurrency and documented the new distributed mode.
Reviewed changes
Copilot reviewed 12 out of 12 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| internal/queue/queue.go | Exposes the queue cache to allow constructing distributed locks from the queue instance. |
| internal/queue/queue_housekeeper.go | Changes unlock processing to run in parallel with configurable concurrency. |
| internal/queue/distributed_lock.go | Introduces distributed lock interface and Redis/NoOp implementations. |
| internal/queue/distributed_lock_test.go | Adds unit tests for the distributed lock implementations. |
| internal/queue/cache/redis_cache.go | Adds Redis operations needed for distributed locking (SetNX, CompareAndDelete, CompareAndExpire, etc.). |
| internal/queue/cache/redis_cache_test.go | Adds Redis integration tests for compare-and-delete and compare-and-expire behavior. |
| internal/queue/cache/redis_cache_scripts.go | Adds Lua scripts for atomic compare-and-delete and compare-and-expire. |
| internal/queue/cache/memory_cache.go | Adds in-memory equivalents of the new cache interface methods (best-effort semantics). |
| internal/queue/cache/cache.go | Extends the cache interface with primitives required for distributed locking. |
| internal/config/housekeeper.go | Adds distributed execution config keys and unlock parallelism config. |
| internal/cmd/deckard/main.go | Updates housekeeper scheduling to acquire/release distributed locks and adds metrics leader logic. |
| docs/distributed-housekeeper.md | Documents the distributed housekeeper feature and configuration. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Introduced a new `lock` package providing a generic distributed mutual-exclusion primitive. - Implemented `Locker` interface with a Redis-backed `storeLocker` for distributed locking. - Added a `noopLocker` for single-instance deployments, preserving original behavior. - Created tests for `noopLocker` and `storeLocker` to ensure correct functionality. - Removed legacy `DistributedLock` implementation and its tests. - Enhanced message structure to include `LockedUntil` timestamp for better lock management. - Updated queue housekeeper logic to handle message recovery based on `LockedUntil`. - Added integration tests to verify Redis lock prefix behavior. - Updated metrics to track housekeeper leadership status.
…related components
- Resolved 13 human review threads - Fixed thread-requested concurrency and lock semantics issues - Updated distributed housekeeper docs to match implementation
Member
Author
PR Review Resolution SummaryPR: feat: implement housekeeper election and leadership (#76) | Branch: copilot/fix-21→main | Last run: 2026-07-05T22:43:00-03:00 | Commit: 8852b88 Threads Processed
SonarQube
CI Checks After Fix
Remaining Open Items
|
- Remove parallelism from tests mutating global dtime provider - Align distributed housekeeper docs with REDIS-only distributed mode - Restore redis cache implementation after rebase regression
Member
Author
PR Review Resolution SummaryPR: feat: implement housekeeper election and leadership (#76) | Branch: copilot/fix-21→main | Last run: 2026-07-06T18:16:00-03:00 | Commit: 11983d3 Threads Processed
SonarQube
CI Checks After Fix
Remaining Open Items
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #76 +/- ##
==========================================
+ Coverage 80.88% 83.23% +2.34%
==========================================
Files 27 30 +3
Lines 3071 3465 +394
==========================================
+ Hits 2484 2884 +400
+ Misses 418 405 -13
- Partials 169 176 +7
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 1 file with indirect coverage changes Continue to review full report in Codecov by Harness.
|
|
|
||
| cache.keys[key] = value | ||
| if ttl > 0 { | ||
| cache.keyExpiresAt[key] = time.Now().Add(ttl) |
| } | ||
|
|
||
| if ttl > 0 { | ||
| cache.keyExpiresAt[key] = time.Now().Add(ttl) |
| } | ||
|
|
||
| if ttl > 0 { | ||
| cache.keyExpiresAt[key] = time.Now().Add(ttl) |
| } | ||
|
|
||
| expiresAt, hasExpiration := cache.keyExpiresAt[key] | ||
| if hasExpiration && !expiresAt.After(time.Now()) { |
Comment on lines
+158
to
+166
| var elector election.Elector | ||
| if distributedExecutionEnabled { | ||
| elector = election.NewLeaseElector(locker, config.HousekeeperElectionLeaseTTL.GetDuration(), instanceID) | ||
| elector.Start(ctx) | ||
| metrics.SetLeaderStatusFunc(elector.IsLeader) | ||
| } else { | ||
| logger.S(ctx).Warnf("Housekeeper distributed execution requires cache type REDIS; running local housekeeper mode with cache type %s", config.CacheType.Get()) | ||
| metrics.SetLeaderStatusFunc(func() bool { return true }) | ||
| } |
Comment on lines
+341
to
+343
| lockName := fmt.Sprintf("housekeeper:lock:%s", taskName) | ||
| lockTTL := config.HousekeeperDistributedExecutionLockTTL.GetDuration() | ||
|
|
Comment on lines
+414
to
416
| lockName := fmt.Sprintf("housekeeper:lock:%s", taskName) | ||
| lockTTL := config.HousekeeperDistributedExecutionLockTTL.GetDuration() | ||
|
|
Comment on lines
+224
to
+228
| Check lock/election keys in Redis: | ||
| ```bash | ||
| redis-cli KEYS "*housekeeper:lock:*" | ||
| redis-cli GET "*housekeeper:election:leader*" | ||
| ``` No newline at end of file |
- internal/config: GetHousekeeperInstanceID (configured/fallback/uniqueness) - internal/queue/cache: MemoryCache Get/Set/Del/Expire/SetNX/CompareAndDelete/ CompareAndExpire TTL and lazy-expiration semantics (real object, no mocks) - internal/lock: TryAcquire expiration and Renew extension through the public Locker API backed by a real MemoryCache - internal/cmd/deckard: startHouseKeeperJobs distributed (Redis leader election + atomic lock/renew cycle) and local mode, exercised directly with short task delays
Tests using go main() with REDIS cache called shutdown.PerformShutdown directly, bypassing main()'s <-shutdown.CancelChan branch that releases housekeeperElector. This leaked the shared Redis leader-election key for up to the configured lease TTL (15s default), causing later tests in the same suite/CI run to fail acquiring leadership within a short timeout. Also flush the cache namespace and extend the leader-election timeout in the new distributed housekeeper test as defense-in-depth against any leftover state from other tests sharing the same Redis instance.
Comment on lines
+104
to
+113
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-e.stopCh: | ||
| return | ||
| case <-ticker.C: | ||
| e.campaign(ctx) | ||
| } | ||
| } |
… tests - internal/queue/cache: RedisCache SetNX/Expire/Close (previously 0%), plus error branches for SetNX/Del/Expire/CompareAndDelete/CompareAndExpire via a genuinely-canceled context (real go-redis error, not mocked) - internal/lock: TryAcquire/Release/Renew error branches exercised through a real RedisCache-backed Store with a canceled context, reaching 100% package coverage (previously required MemoryCache, which never returns errors) - internal/metrics: SetLeaderStatusFunc and metrifyLeaderStatus exercised end-to-end by scraping the real Prometheus registry for the deckard_housekeeper_leader gauge value - internal/election: LeaseElector demotes itself when lease renewal fails (external key loss without Stop()), covering campaign()'s lost-leadership branch Whole-repo coverage: 73.0% -> 74.3% (main is 74.2%), 511 tests, 0 failures.
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
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.
Fixes #21
Partial fix for #30