feat: replace sidekiq reaper+recovery with atomic ClaimSidekiqJob, per-owner heartbeat fencing#5000
Conversation
📝 WalkthroughWalkthroughThis PR switches Sidekiq job handling to owner-fenced claiming and heartbeat-based execution, adds the owner_id schema change, updates store and runner APIs, and rewires server startup/shutdown around a dispatcher instead of reaper/recovery logic. ChangesSidekiq ownership claim/heartbeat model
Estimated code review effort: 4 (Complex) | ~75 minutes Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
|
Warning This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
This stack of pull requests is managed by Graphite. Learn more about stacking. |
BearTS
changed the title
ClaimSidekiqJob, per-owner heartbeat fencing, and cluster-safe dispatcher
BearTS
changed the title
ClaimSidekiqJob, per-owner heartbeat fencing, and cluster-safe dispatcherClaimSidekiqJob, per-owner heartbeat fencing
Confidence Score: 4/5Core cluster-safety logic is correct and well-tested; two open items from earlier review rounds remain unresolved before merging. The atomic claim, owner fencing, heartbeat goroutine, and inflight dedup are all implemented correctly. Previous concerns about the staleAfter data race and unbounded goroutine accumulation are addressed in this diff. Two items flagged in prior rounds are still outstanding: the migration index on the sidekiq table is created with a blocking statement rather than CONCURRENTLY, which can lock the table on deployments with accumulated history; and TestConcurrencySemaphoreBound has a WaitGroup counter underflow that causes the third job to land in failed rather than completed, making the test assert incorrect runtime behaviour while appearing to pass. framework/configstore/migrations.go (blocking index creation) and framework/sidekiq/sidekiq_test.go (WaitGroup underflow in TestConcurrencySemaphoreBound) Important Files Changed
Sequence Diagram%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant D as StartDispatcher (ticker)
participant Store as ConfigStore (DB)
participant E as execute goroutine
participant H as Handler
participant HB as Heartbeat goroutine
D->>Store: ListClaimableSidekiqJobs(staleBefore)
Store-->>D: [pending jobs, stale-running jobs]
D->>E: spawn(job) [if not inflight]
note over E: acquire semaphore
E->>Store: ClaimSidekiqJob(id, ownerID, staleBefore)
note over Store: atomic UPDATE WHERE status=pending OR (status=running AND updated_at lt staleBefore)
Store-->>E: "claimed=true (RowsAffected==1)"
E->>HB: startHeartbeat(jobCtx, cancel, id)
E->>H: fn(jobCtx, job, progress)
loop every HeartbeatInterval
HB->>Store: HeartbeatSidekiqJob(id, ownerID)
Store-->>HB: "ok=true (still owns)"
end
H-->>E: finalMetadata, nil
E->>HB: stopHeartbeat()
E->>Store: CompleteSidekiqJob(id, ownerID, metadata)
note over Store: fenced: WHERE owner_id=ownerID AND status=running
note over HB: ownership lost scenario
HB->>Store: HeartbeatSidekiqJob(id, ownerID)
Store-->>HB: "ok=false (owner changed)"
HB->>E: cancel() cancels jobCtx
H-->>E: context.Canceled error
E->>Store: FailSidekiqJob(id, ownerID, ...) returns error: not owner
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant D as StartDispatcher (ticker)
participant Store as ConfigStore (DB)
participant E as execute goroutine
participant H as Handler
participant HB as Heartbeat goroutine
D->>Store: ListClaimableSidekiqJobs(staleBefore)
Store-->>D: [pending jobs, stale-running jobs]
D->>E: spawn(job) [if not inflight]
note over E: acquire semaphore
E->>Store: ClaimSidekiqJob(id, ownerID, staleBefore)
note over Store: atomic UPDATE WHERE status=pending OR (status=running AND updated_at lt staleBefore)
Store-->>E: "claimed=true (RowsAffected==1)"
E->>HB: startHeartbeat(jobCtx, cancel, id)
E->>H: fn(jobCtx, job, progress)
loop every HeartbeatInterval
HB->>Store: HeartbeatSidekiqJob(id, ownerID)
Store-->>HB: "ok=true (still owns)"
end
H-->>E: finalMetadata, nil
E->>HB: stopHeartbeat()
E->>Store: CompleteSidekiqJob(id, ownerID, metadata)
note over Store: fenced: WHERE owner_id=ownerID AND status=running
note over HB: ownership lost scenario
HB->>Store: HeartbeatSidekiqJob(id, ownerID)
Store-->>HB: "ok=false (owner changed)"
HB->>E: cancel() cancels jobCtx
H-->>E: context.Canceled error
E->>Store: FailSidekiqJob(id, ownerID, ...) returns error: not owner
Reviews (4): Last reviewed commit: "feat: respect sidekiq over clustering" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| tx = tx.WithContext(ctx) | ||
| if err := tx.Exec(`DROP INDEX IF EXISTS idx_sidekiq_owner`).Error; err != nil { | ||
| return err | ||
| } | ||
| return dropColumnIfExists(tx, logger, &tables.TableSidekiqJob{}, "owner_id") | ||
| }, | ||
| }}) | ||
| if err := m.Migrate(); err != nil { | ||
| return fmt.Errorf("error running %s migration: %w", migrationName, err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
CREATE INDEX IF NOT EXISTS idx_sidekiq_owner ON sidekiq (owner_id) acquires an AccessShareLock-conflicting write lock for the entire index build. For a PostgreSQL deployment with accumulated historical job rows, this blocks all concurrent reads and writes on the sidekiq table for the duration of the migration. Per the migration-large-table-safety rule, index creation must use CREATE INDEX CONCURRENTLY where the database supports it.
For SQLite there is no CONCURRENTLY keyword, so the migration should branch on the dialect (or use a helper that already does so), or document that this migration requires a maintenance window if the table is large. The same concern applies to the Rollback's DROP INDEX — it should use DROP INDEX CONCURRENTLY IF EXISTS on PostgreSQL.
Rule Used: Whenever a migration is added or changed, verify i... (source)
| // Poison guard: job.Attempts is the count before this claim. Once a job has been | ||
| // attempted MaxAttempts times, stop retrying and mark it failed permanently. | ||
| if job.Attempts >= MaxAttempts { | ||
| r.logger.Warn("sidekiq: job %s (%s) exceeded max attempts (%d), marking failed", job.ID, job.Kind, MaxAttempts) | ||
| if ferr := r.store.FailSidekiqJob(r.baseCtx, job.ID, r.ownerID, job.Metadata, fmt.Sprintf("exceeded max attempts (%d)", MaxAttempts)); ferr != nil { | ||
| r.logger.Error("sidekiq: failed to fail exhausted job %s: %v", job.ID, ferr) | ||
| } | ||
| return | ||
| } |
There was a problem hiding this comment.
MaxAttempts check uses the pre-claim snapshot — no test covers the boundary
job.Attempts is the value captured when the job was listed or enqueued; the ClaimSidekiqJob call on line 184 has already incremented the DB counter by the time this guard runs. When job.Attempts == MaxAttempts - 1 in the snapshot, the 5th run is allowed (correct); when job.Attempts == MaxAttempts, the guard fires and the job is permanently failed (correct). The logic is right, but there is no dedicated test that exercises this path. A test seeding a job with Attempts = MaxAttempts and verifying that execute() calls FailSidekiqJob rather than the handler would prevent regressions here.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
@CodeRabbit review |
✅ Action performedReview finished.
|
BearTS
force-pushed
the
07-07-feat_respect_sidekiq_over_clustering
branch
from
8e175f9 to
9db4367
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
BearTS
force-pushed
the
07-07-feat_respect_sidekiq_over_clustering
branch
from
9db4367 to
91634f5
Compare
BearTS
force-pushed
the
07-07-feat_add_sidekiq_job_runner
branch
from
6257422 to
e7bfb8f
Compare
BearTS
force-pushed
the
07-07-feat_respect_sidekiq_over_clustering
branch
from
91634f5 to
c721fd7
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
framework/sidekiq/sidekiq.go (1)
180-199: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBound goroutines, not just active handlers.
dispatchOncecan list all claimable jobs andspawnstarts one goroutine per job before acquiringsem. With a large backlog, this can create an unbounded number of goroutines blocked on the semaphore; acquire a slot before launching or add a bounded dispatch batch/queue.One localized way to bound spawned goroutines
func (r *Runner) spawn(job tables.TableSidekiqJob) { if !r.tryMarkInflight(job.ID) { return } + select { + case r.sem <- struct{}{}: + case <-r.baseCtx.Done(): + r.clearInflight(job.ID) + return + default: + r.clearInflight(job.ID) + return + } r.wg.Add(1) go func() { defer r.wg.Done() defer r.clearInflight(job.ID) - select { - case r.sem <- struct{}{}: - case <-r.baseCtx.Done(): - return - } defer func() { <-r.sem }() r.execute(job) }() }As per coding guidelines, Go concurrency should use bounded goroutines/channels; as per path instructions,
framework/**changes should preserve bounded resource usage.Also applies to: 346-357
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@framework/sidekiq/sidekiq.go` around lines 180 - 199, The goroutine spawning in Runner.spawn is still unbounded because each claimable job launches a goroutine before acquiring sem, which can pile up blocked goroutines under backlog. Fix this by ensuring concurrency is bounded before goroutine creation, either by acquiring a semaphore slot in dispatchOnce/spawn prior to starting the goroutine or by introducing a bounded dispatch batch/queue so only a limited number of jobs are handed off at once. Keep the behavior centered around Runner.spawn and dispatchOnce so the node never accumulates more goroutines than the configured concurrency.Sources: Coding guidelines, Path instructions
framework/configstore/sidekiq.go (1)
108-123: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winFence progress writes on
status = runningtoo.MarkStaleSidekiqJobsFailedcan flip a running job tofailedwithout changingowner_id, so the previous owner can still callUpdateSidekiqJobProgressand overwritemetadata/updated_aton a terminal row. MatchCompleteSidekiqJobandFailSidekiqJobhere to keep stale jobs from being updated after the reaper has failed them.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@framework/configstore/sidekiq.go` around lines 108 - 123, Update UpdateSidekiqJobProgress in RDBConfigStore so progress writes are gated on the job still being running, not just matching id and owner_id. Add the same status check used by CompleteSidekiqJob and FailSidekiqJob to the query before calling Updates, so MarkStaleSidekiqJobsFailed cannot leave a failed row writable by the previous owner. Keep the existing not-found/ownership error path, but ensure terminal jobs are excluded from metadata and updated_at changes.
🧹 Nitpick comments (3)
framework/configstore/tables/sidekiq.go (1)
29-29: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winName the GORM index tag to match the migration's index name.
OwnerID'sindextag is unnamed, while the raw-SQL migration creates a specifically namedidx_sidekiq_ownerindex. Other fields in this struct (Kind,Status,UpdatedAt) explicitly name their index tags. If GORM's own naming convention is ever exercised (e.g. via thedefault:dialect fallback inmigrationAddSidekiqTablethat callstx.Migrator().CreateTable), this mismatch could produce a duplicate, differently-named index on the same column.♻️ Proposed fix
- OwnerID string `gorm:"column:owner_id;type:text;index" json:"owner_id,omitempty"` + OwnerID string `gorm:"column:owner_id;type:text;index:idx_sidekiq_owner" json:"owner_id,omitempty"`🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@framework/configstore/tables/sidekiq.go` at line 29, The OwnerID field in SidekiqTable uses an unnamed GORM index tag, which can diverge from the migration’s explicit idx_sidekiq_owner index name. Update the OwnerID struct tag in SidekiqTable to use the same named index pattern as Kind, Status, and UpdatedAt so migrationAddSidekiqTable and tx.Migrator().CreateTable produce consistent index naming and avoid duplicates.framework/configstore/store.go (1)
666-675: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument the ownership/fencing contract for the new Sidekiq methods.
ClaimSidekiqJob,HeartbeatSidekiqJob, and the owner-fencedUpdateSidekiqJobProgress/CompleteSidekiqJob/FailSidekiqJobintroduce a new atomic-claim + fencing contract, but unlike other concurrency-sensitive interface methods in this file (e.g.TryAcquireLock,ConsentOAuth2AuthorizeRequest), they lack doc comments describing what theboolreturn means (claim won vs. lost), what "stale" means relative tostaleBefore, and whether an ownership mismatch on the mutating methods is a silent no-op or returns an error. This matters especially since this contract spans multiple layers (RDB implementation, runner, dispatcher) per the PR's stacked design.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@framework/configstore/store.go` around lines 666 - 675, Document the new Sidekiq ownership/fencing contract on the interface methods in Store: add concise comments for ClaimSidekiqJob, HeartbeatSidekiqJob, UpdateSidekiqJobProgress, CompleteSidekiqJob, and FailSidekiqJob describing what the bool return means, how staleBefore defines staleness, and how ownership mismatches are handled. Mirror the style used by other concurrency-sensitive methods like TryAcquireLock and ConsentOAuth2AuthorizeRequest, and make sure the comments clearly state the atomic-claim plus fenced-owner behavior that the RDB implementation, runner, and dispatcher rely on.framework/sidekiq/sidekiq.go (1)
135-136: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAvoid implying
Enqueuealways wins the claim.A dispatcher on another node can claim the pending row between
CreateSidekiqJoband this node’sexecute; the atomic claim decides the winner.Suggested comment adjustment
-// spawns the job directly and wins the claim (the row is pending and unowned). +// spawns the job directly; the first runner to atomically claim the pending row +// processes it, so a dispatcher on another node may still win the race.As per coding guidelines, Go changes should preserve clear ownership semantics.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@framework/sidekiq/sidekiq.go` around lines 135 - 136, Adjust the comment in the sidekiq enqueue flow to avoid saying the enqueuing node “wins the claim” unconditionally; update the wording near CreateSidekiqJob and execute to reflect that the atomic claim determines ownership and another dispatcher can claim the pending row before this node executes. Keep the semantics clear that Enqueue only creates the job and background execution/claim ownership is decided later.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@framework/configstore/sidekiq_test.go`:
- Around line 177-190: Update UpdateSidekiqJobProgress to only write progress
when the job is still in the running state, not just when owner_id matches. Add
the status=f.running fence in the Sidekiq job update path so completed or failed
jobs cannot have metadata overwritten, and keep the owner check intact. Extend
TestUpdateSidekiqJobProgress in sidekiq_test.go to cover a terminal-state case
by asserting progress updates are rejected once the job is no longer running,
using the existing create/claim/update flow helpers.
In `@framework/sidekiq/sidekiq_test.go`:
- Around line 449-454: The handler registered via r.Register("k", ...) is
calling inFlight.Done() for every job, but inFlight.Add(maxConcurrent) only
covers the first two expected executions, so the extra job can drive the
WaitGroup negative. Update the test’s synchronization in sidekiq_test.go so the
"k" handler only signals the WaitGroup for the intended in-flight jobs (or
conditionally skips Done() for job-extra), while still blocking on gate and
preserving the concurrent job sequencing.
- Around line 111-121: The fake progress update path in
fakeStore.UpdateSidekiqJobProgress only checks ownership, so it still accepts
checkpoints after a job has left running state. Update this method to also
verify the job is still running before mutating metadata/progress, matching the
terminal-write guard used elsewhere in the Sidekiq fake. Keep the existing
ownership check and reject late progress updates for completed/failed/reaped
jobs so runner tests exercise the failure path correctly.
In `@transports/bifrost-http/server/server.go`:
- Around line 1901-1905: The server error exit path in Server.Start leaves the
Sidekiq dispatcher running, so clean up the background worker on every Start
return path. Update the logic around SidekiqRunner.StartDispatcher in
Server.Start to ensure SidekiqDispatcherStop is called and
SidekiqRunner.Shutdown is invoked before returning on any error after the
dispatcher has been started. Make the cleanup safe to call only when
SidekiqRunner is non-nil, and preserve the existing lifecycle/cancellation
behavior for the dispatcher and HTTP server shutdown flow.
---
Outside diff comments:
In `@framework/configstore/sidekiq.go`:
- Around line 108-123: Update UpdateSidekiqJobProgress in RDBConfigStore so
progress writes are gated on the job still being running, not just matching id
and owner_id. Add the same status check used by CompleteSidekiqJob and
FailSidekiqJob to the query before calling Updates, so
MarkStaleSidekiqJobsFailed cannot leave a failed row writable by the previous
owner. Keep the existing not-found/ownership error path, but ensure terminal
jobs are excluded from metadata and updated_at changes.
In `@framework/sidekiq/sidekiq.go`:
- Around line 180-199: The goroutine spawning in Runner.spawn is still unbounded
because each claimable job launches a goroutine before acquiring sem, which can
pile up blocked goroutines under backlog. Fix this by ensuring concurrency is
bounded before goroutine creation, either by acquiring a semaphore slot in
dispatchOnce/spawn prior to starting the goroutine or by introducing a bounded
dispatch batch/queue so only a limited number of jobs are handed off at once.
Keep the behavior centered around Runner.spawn and dispatchOnce so the node
never accumulates more goroutines than the configured concurrency.
---
Nitpick comments:
In `@framework/configstore/store.go`:
- Around line 666-675: Document the new Sidekiq ownership/fencing contract on
the interface methods in Store: add concise comments for ClaimSidekiqJob,
HeartbeatSidekiqJob, UpdateSidekiqJobProgress, CompleteSidekiqJob, and
FailSidekiqJob describing what the bool return means, how staleBefore defines
staleness, and how ownership mismatches are handled. Mirror the style used by
other concurrency-sensitive methods like TryAcquireLock and
ConsentOAuth2AuthorizeRequest, and make sure the comments clearly state the
atomic-claim plus fenced-owner behavior that the RDB implementation, runner, and
dispatcher rely on.
In `@framework/configstore/tables/sidekiq.go`:
- Line 29: The OwnerID field in SidekiqTable uses an unnamed GORM index tag,
which can diverge from the migration’s explicit idx_sidekiq_owner index name.
Update the OwnerID struct tag in SidekiqTable to use the same named index
pattern as Kind, Status, and UpdatedAt so migrationAddSidekiqTable and
tx.Migrator().CreateTable produce consistent index naming and avoid duplicates.
In `@framework/sidekiq/sidekiq.go`:
- Around line 135-136: Adjust the comment in the sidekiq enqueue flow to avoid
saying the enqueuing node “wins the claim” unconditionally; update the wording
near CreateSidekiqJob and execute to reflect that the atomic claim determines
ownership and another dispatcher can claim the pending row before this node
executes. Keep the semantics clear that Enqueue only creates the job and
background execution/claim ownership is decided later.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 4c53b470-6b2a-45fb-8d64-1660e9044ffa
📒 Files selected for processing (9)
framework/configstore/migrations.goframework/configstore/sidekiq.goframework/configstore/sidekiq_test.goframework/configstore/store.goframework/configstore/tables/sidekiq.goframework/sidekiq/sidekiq.goframework/sidekiq/sidekiq_test.gotransports/bifrost-http/lib/config_test.gotransports/bifrost-http/server/server.go
| func TestUpdateSidekiqJobProgress(t *testing.T) { | ||
| store := setupSidekiqTestStore(t) | ||
| ctx := context.Background() | ||
| require.NoError(t, store.CreateSidekiqJob(ctx, &tables.TableSidekiqJob{ID: "j1", Kind: "k"})) | ||
| _, err := store.ClaimSidekiqJob(ctx, "j1", "owner-A", time.Now().Add(-time.Minute)) | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, store.UpdateSidekiqJobProgress(ctx, "j1", "owner-A", `{"cursor":42}`)) | ||
| assert.Equal(t, `{"cursor":42}`, getJob(t, store, "j1").Metadata) | ||
|
|
||
| // A stale/non-owner cannot advance progress. | ||
| assert.Error(t, store.UpdateSidekiqJobProgress(ctx, "j1", "owner-B", `{"cursor":99}`)) | ||
| assert.Equal(t, `{"cursor":42}`, getJob(t, store, "j1").Metadata, "metadata unchanged by non-owner") | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map relevant files first
git ls-files framework/configstore | sed -n '1,200p'
echo '---'
ast-grep outline framework/configstore/sidekiq_test.go --view expanded
echo '---'
ast-grep outline framework/configstore | sed -n '1,200p'
# Locate the implementation of the progress update path and terminal-state transitions
echo '--- rg UpdateSidekiqJobProgress/CompleteSidekiqJob/Fail/Reap ---'
rg -n "UpdateSidekiqJobProgress|CompleteSidekiqJob|FailSidekiqJob|ReapSidekiqJob|ClaimSidekiqJob" framework/configstore -S
# Show the implementation around the progress update logic and terminal updates
echo '--- file sizes ---'
wc -l framework/configstore/*.go framework/configstore/*_test.go || true
echo '--- sidekiq implementation slice ---'
sed -n '1,260p' framework/configstore/sidekiq.go
echo '--- sidekiq test slice ---'
sed -n '1,260p' framework/configstore/sidekiq_test.go
# Read only the specific sections if the files are large
echo '--- focused impl slices ---'
sed -n '1,220p' framework/configstore/sidekiq.go | nl -ba | sed -n '1,220p'Repository: maximhq/bifrost
Length of output: 50373
Guard progress updates on terminal jobs. UpdateSidekiqJobProgress only checks owner_id, so the same owner can still overwrite metadata after the job has been completed or failed. Add a status = running fence here and cover the terminal-state case in framework/configstore/sidekiq_test.go.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@framework/configstore/sidekiq_test.go` around lines 177 - 190, Update
UpdateSidekiqJobProgress to only write progress when the job is still in the
running state, not just when owner_id matches. Add the status=f.running fence in
the Sidekiq job update path so completed or failed jobs cannot have metadata
overwritten, and keep the owner check intact. Extend
TestUpdateSidekiqJobProgress in sidekiq_test.go to cover a terminal-state case
by asserting progress updates are rejected once the job is no longer running,
using the existing create/claim/update flow helpers.
Source: Path instructions
| func (f *fakeStore) UpdateSidekiqJobProgress(_ context.Context, id, ownerID, metadata string) error { | ||
| f.mu.Lock() | ||
| defer f.mu.Unlock() | ||
| j, ok := f.jobs[id] | ||
| if !ok || j.owner != ownerID { | ||
| return errors.New("not owned by caller") | ||
| } | ||
| j.metadata = metadata | ||
| j.updatedAt = time.Now() | ||
| f.progress[id] = metadata | ||
| return nil |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fence fake progress updates on running status too.
The fake currently allows the same owner to checkpoint a completed/failed job, so runner tests can miss late-progress mutations after terminal/reaped state. Match the terminal-write guard and reject progress unless the job is still running.
Suggested fix
j, ok := f.jobs[id]
- if !ok || j.owner != ownerID {
- return errors.New("not owned by caller")
+ if !ok || j.owner != ownerID || j.status != tables.SidekiqStatusRunning {
+ return errors.New("not owned by caller or no longer running")
}As per path instructions, framework tests should cover failure paths and preserve data ownership semantics.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (f *fakeStore) UpdateSidekiqJobProgress(_ context.Context, id, ownerID, metadata string) error { | |
| f.mu.Lock() | |
| defer f.mu.Unlock() | |
| j, ok := f.jobs[id] | |
| if !ok || j.owner != ownerID { | |
| return errors.New("not owned by caller") | |
| } | |
| j.metadata = metadata | |
| j.updatedAt = time.Now() | |
| f.progress[id] = metadata | |
| return nil | |
| func (f *fakeStore) UpdateSidekiqJobProgress(_ context.Context, id, ownerID, metadata string) error { | |
| f.mu.Lock() | |
| defer f.mu.Unlock() | |
| j, ok := f.jobs[id] | |
| if !ok || j.owner != ownerID || j.status != tables.SidekiqStatusRunning { | |
| return errors.New("not owned by caller or no longer running") | |
| } | |
| j.metadata = metadata | |
| j.updatedAt = time.Now() | |
| f.progress[id] = metadata | |
| return nil | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@framework/sidekiq/sidekiq_test.go` around lines 111 - 121, The fake progress
update path in fakeStore.UpdateSidekiqJobProgress only checks ownership, so it
still accepts checkpoints after a job has left running state. Update this method
to also verify the job is still running before mutating metadata/progress,
matching the terminal-write guard used elsewhere in the Sidekiq fake. Keep the
existing ownership check and reject late progress updates for
completed/failed/reaped jobs so runner tests exercise the failure path
correctly.
Source: Path instructions
| var inFlight sync.WaitGroup | ||
|
|
||
| r.Register("k", func(_ context.Context, _ tables.TableSidekiqJob, _ ProgressFunc) (string, error) { | ||
| inFlight.Done() // signal that we entered the handler | ||
| <-gate // block until the test releases us | ||
| return "", nil |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Avoid a negative WaitGroup when the extra job runs.
inFlight.Add(maxConcurrent) accounts for only the first two handlers, but the handler calls inFlight.Done() for job-extra as well. Once the gate is released, the extra handler can drive the counter negative and panic or be recovered as a failed job.
Suggested fix
- r.Register("k", func(_ context.Context, _ tables.TableSidekiqJob, _ ProgressFunc) (string, error) {
- inFlight.Done() // signal that we entered the handler
- <-gate // block until the test releases us
+ r.Register("k", func(_ context.Context, job tables.TableSidekiqJob, _ ProgressFunc) (string, error) {
+ if job.ID != "job-extra" {
+ inFlight.Done() // signal that the first batch entered the handler
+ }
+ <-gate // block until the test releases us
return "", nil
})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| var inFlight sync.WaitGroup | |
| r.Register("k", func(_ context.Context, _ tables.TableSidekiqJob, _ ProgressFunc) (string, error) { | |
| inFlight.Done() // signal that we entered the handler | |
| <-gate // block until the test releases us | |
| return "", nil | |
| var inFlight sync.WaitGroup | |
| r.Register("k", func(_ context.Context, job tables.TableSidekiqJob, _ ProgressFunc) (string, error) { | |
| if job.ID != "job-extra" { | |
| inFlight.Done() // signal that the first batch entered the handler | |
| } | |
| <-gate // block until the test releases us | |
| return "", nil |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@framework/sidekiq/sidekiq_test.go` around lines 449 - 454, The handler
registered via r.Register("k", ...) is calling inFlight.Done() for every job,
but inFlight.Add(maxConcurrent) only covers the first two expected executions,
so the extra job can drive the WaitGroup negative. Update the test’s
synchronization in sidekiq_test.go so the "k" handler only signals the WaitGroup
for the intended in-flight jobs (or conditionally skips Done() for job-extra),
while still blocking on gate and preserving the concurrent job sequencing.
| // Start the Sidekiq dispatcher: on every node it periodically claims pending and | ||
| // stale (orphaned) jobs, with an atomic claim guaranteeing exactly one node runs | ||
| // each job. Subsumes startup recovery of jobs left behind by a crash or restart. | ||
| if s.SidekiqRunner != nil { | ||
| s.SidekiqReaperStop = s.SidekiqRunner.StartReaper(sidekiq.ReaperInterval, sidekiq.StaleAfter) | ||
| go func() { | ||
| if err := s.SidekiqRunner.RecoverIncomplete(ctx); err != nil { | ||
| logger.Error("sidekiq: failed to recover incomplete provisioning jobs: %v", err) | ||
| } | ||
| }() | ||
| s.SidekiqDispatcherStop = s.SidekiqRunner.StartDispatcher(sidekiq.DispatchInterval, sidekiq.StaleAfter) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Stop the dispatcher on every Start exit path.
Line 1905 starts a dispatcher goroutine that immediately scans and later claims jobs, but the server-error branch at Lines 2024-2031 returns without calling SidekiqDispatcherStop() or SidekiqRunner.Shutdown(). If Serve exits with an error while the process continues, this node can keep claiming Sidekiq jobs after the HTTP server is down.
Proposed cleanup for the server-error path
case err := <-errChan:
if s.IntegrationHandler != nil {
s.IntegrationHandler.Close()
}
+ if s.cancel != nil {
+ s.cancel()
+ }
+ if s.SidekiqDispatcherStop != nil {
+ logger.Info("stopping sidekiq dispatcher...")
+ s.SidekiqDispatcherStop()
+ }
+ if s.SidekiqRunner != nil {
+ logger.Info("stopping sidekiq runner...")
+ s.SidekiqRunner.Shutdown()
+ }
if s.wsPool != nil {
s.wsPool.Close()
}
return errAs per path instructions, transports changes should preserve lifecycle behavior and Go review should check correct context cancellation and bounded background resources.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@transports/bifrost-http/server/server.go` around lines 1901 - 1905, The
server error exit path in Server.Start leaves the Sidekiq dispatcher running, so
clean up the background worker on every Start return path. Update the logic
around SidekiqRunner.StartDispatcher in Server.Start to ensure
SidekiqDispatcherStop is called and SidekiqRunner.Shutdown is invoked before
returning on any error after the dispatcher has been started. Make the cleanup
safe to call only when SidekiqRunner is non-nil, and preserve the existing
lifecycle/cancellation behavior for the dispatcher and HTTP server shutdown
flow.
Source: Path instructions
BearTS
changed the base branch from
07-07-feat_add_sidekiq_job_runner
to
graphite-base/5000
1 participant
Summary
Replaces the single-node Sidekiq reaper +
RecoverIncompletepattern with a cluster-safe dispatcher backed by atomic job claiming. Previously, multiple nodes could double-run the same job on startup recovery, and a revived stale owner could overwrite state written by the node that re-claimed the job. This PR introduces anowner_idcolumn and per-operation owner fencing so exactly one node runs each job at any time, even across crashes and restarts.Changes
owner_idcolumn and migration: AddedOwnerIDtoTableSidekiqJoband a new migration (add_sidekiq_owner_id) that adds the column and an index. All mutating operations (progress,complete,fail,heartbeat) are now fenced onowner_idso a revived stale owner affects 0 rows.ClaimSidekiqJobreplacesMarkSidekiqJobRunning: The new method is an atomic conditionalUPDATEthat succeeds only when the job is pending, or running but stale (heartbeat older thanstaleBefore). A job with a live owner yieldsRowsAffected == 0.started_atis set viaCOALESCEso resumes preserve the original start time. Returns a boolean indicating whether this caller won the claim.HeartbeatSidekiqJob: New method that bumpsupdated_atfor a job the caller still owns. Returnsfalsewhen ownership has been lost, which the runner uses as a signal to cancel the in-flight handler context.ListClaimableSidekiqJobsreplacesListIncompleteSidekiqJobs: Now filters to pending jobs and running-but-stale jobs (orphaned by a dead owner), rather than all non-terminal jobs.Per-job heartbeat goroutine: Each claimed job runs a ticker that periodically calls
HeartbeatSidekiqJob. If the heartbeat reports lost ownership, the job's context is cancelled so the handler stops at its next checkpoint.StartDispatcherreplacesStartReaper+RecoverIncomplete: Runs one scan immediately on startup (subsuming crash recovery) then on a fixed interval. Each node lists claimable jobs and spawns goroutines; the atomic claim insideexecuteensures exactly one node wins per job — no leader election required.Poison-job guard: Jobs that have reached
MaxAttempts(5) are marked failed permanently rather than retried indefinitely.ownerIDper process:Runnergenerates a UUID on construction that identifies it for the lifetime of the process. All store calls pass this ID for fencing.Tests updated:
fakeStorenow models atomic claim and owner fencing. New tests cover: dispatcher picking up pending and stale jobs, claim exclusivity across two owners, single-winner race between two runners sharing a store, and heartbeat-triggered handler cancellation on lost ownership.Type of change
Affected areas
How to test
Key scenarios to validate:
TestRunnerRaceSingleWinner).TestDispatcherRunsClaimableJobs).TestHeartbeatCancelsOnLostOwnership).add_sidekiq_owner_idmigration applies cleanly on a database that already has thesidekiqtable.Breaking changes
The
ConfigStoreinterface has changed:MarkSidekiqJobRunning,ListIncompleteSidekiqJobsare removed;ClaimSidekiqJob,HeartbeatSidekiqJob,ListClaimableSidekiqJobsare added;UpdateSidekiqJobProgress,CompleteSidekiqJob, andFailSidekiqJobnow require anownerIDargument. Any external implementation ofConfigStoreorsidekiq.Storemust be updated.BifrostHTTPServer.SidekiqReaperStopis renamed toSidekiqDispatcherStop.Security considerations
owner_idis a process-scoped UUID generated at startup. It is stored in the database but is not user-facing and carries no elevated privilege. The fencing it provides prevents a compromised or stale node from overwriting job state owned by a healthy node.Checklist
docs/contributing/README.mdand followed the guidelines