Skip to content

test: [#1006] wait for the worker to stop in the queue worker tests #1566

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hwbrzzl merged 1 commit into from
Sep 28, 2026
Merged

test: [#1006] wait for the worker to stop in the queue worker tests #1566

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 65 additions & 98 deletions queue/worker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,14 +339,11 @@ func (s *WorkerTestSuite) Test_run() {
s.SetupTest()
s.mockDriver.EXPECT().Pop(queue).Return(nil, errors.QueueDriverNoJobFound).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("failed to pop job", func() {
Expand All @@ -355,14 +352,11 @@ func (s *WorkerTestSuite) Test_run() {

s.mockLog.EXPECT().Error(errors.QueueDriverFailedToPop.Args(queue, assert.AnError)).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("job failed, insert failed job", func() {
Expand Down Expand Up @@ -390,14 +384,11 @@ func (s *WorkerTestSuite) Test_run() {
s.mockDB.EXPECT().Table("failed_jobs").Return(mockQuery).Once()
mockQuery.EXPECT().Insert(failedJob).Return(nil, nil).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("failed to insert failed job", func() {
Expand Down Expand Up @@ -426,14 +417,11 @@ func (s *WorkerTestSuite) Test_run() {
mockQuery.EXPECT().Insert(failedJob).Return(nil, assert.AnError).Once()
s.mockLog.EXPECT().Error(errors.QueueFailedToSaveFailedJob.Args(assert.AnError, *failedJob)).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("chain job failed, insert failed job", func() {
Expand Down Expand Up @@ -498,14 +486,11 @@ func (s *WorkerTestSuite) Test_run() {
FailedAt: carbon.NewDateTime(carbon.Now()),
}).Return(nil, nil).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("happy path", func() {
Expand Down Expand Up @@ -533,14 +518,11 @@ func (s *WorkerTestSuite) Test_run() {
mockReservedJob.EXPECT().Delete().Return(nil).Once()
s.mockDriver.EXPECT().Pop(queue).Return(nil, errors.QueueDriverNoJobFound).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(1500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("happy path with chain", func() {
Expand Down Expand Up @@ -572,14 +554,11 @@ func (s *WorkerTestSuite) Test_run() {
mockReservedJob.EXPECT().Delete().Return(nil).Once()
s.mockDriver.EXPECT().Pop(queue).Return(nil, errors.QueueDriverNoJobFound).Once()

go func() {
err := s.worker.run()
s.NoError(err)
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("retry job failed, insert failed job", func() {
Expand Down Expand Up @@ -644,15 +623,12 @@ func (s *WorkerTestSuite) Test_run() {
FailedAt: carbon.NewDateTime(carbon.Now()),
}).Return(nil, nil).Once()

go func() {
s.worker.tries = 2 // Set tries to 2 for retry
err := s.worker.run()
s.NoError(err)
}()
s.worker.tries = 2 // Set tries to 2 for retry
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
s.worker.tries = 1
})

Expand Down Expand Up @@ -696,15 +672,12 @@ func (s *WorkerTestSuite) Test_run() {
mockReservedJob.EXPECT().Delete().Return(nil).Once()
s.mockDriver.EXPECT().Pop(queue).Return(nil, errors.QueueDriverNoJobFound)

go func() {
s.worker.tries = 3 // Set tries to 3 for retry
err := s.worker.run()
s.NoError(err)
}()
s.worker.tries = 3 // Set tries to 3 for retry
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)

s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
s.worker.tries = 1
})
}
Expand Down Expand Up @@ -759,12 +732,10 @@ func (s *WorkerTestSuite) Test_runWithReceive() {
return nil, ctx.Err()
}).Once()

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

time.Sleep(200 * time.Millisecond)
s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("receive error", func() {
Expand All @@ -788,12 +759,10 @@ func (s *WorkerTestSuite) Test_runWithReceive() {

s.mockLog.EXPECT().Error(errors.QueueDriverFailedToReceive.Args(queue, assert.AnError)).Once()

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)
s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("happy path", func() {
Expand Down Expand Up @@ -828,12 +797,10 @@ func (s *WorkerTestSuite) Test_runWithReceive() {

s.mockJob.EXPECT().Call(successTask.Job.Signature(), make([]any, 0)).Return(nil).Once()

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)
s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("job failed, insert failed job", func() {
Expand Down Expand Up @@ -881,12 +848,10 @@ func (s *WorkerTestSuite) Test_runWithReceive() {
s.mockDB.EXPECT().Table("failed_jobs").Return(mockQuery).Once()
mockQuery.EXPECT().Insert(matchFailedJob("test-receive-err")).Return(nil, nil).Once()

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)
s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("batch messages", func() {
Expand Down Expand Up @@ -925,12 +890,10 @@ func (s *WorkerTestSuite) Test_runWithReceive() {

s.mockJob.EXPECT().Call(testJobOne.Signature(), make([]any, 0)).Return(nil).Times(3)

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

time.Sleep(500 * time.Millisecond)
s.NoError(s.worker.Shutdown())
s.shutdownWorker(runErrChan)
})

s.Run("shutdown waits for receive loop", func() {
Expand Down Expand Up @@ -962,16 +925,13 @@ func (s *WorkerTestSuite) Test_runWithReceive() {
mockDriverWithReceive.EXPECT().Receive(mock.Anything, queue, s.worker.concurrent).
Return([]contractsqueue.ReservedJob{mockReservedJob}, nil).Once()

go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Shutdown is called while job is still running (job takes 300ms, shutdown at 100ms)
time.Sleep(100 * time.Millisecond)

// Shutdown should block until the in-flight job completes
err := s.worker.Shutdown()
s.NoError(err)
s.shutdownWorker(runErrChan)
s.True(called)
})
}
Expand Down Expand Up @@ -1000,17 +960,13 @@ func (s *WorkerTestSuite) TestShutdown() {
s.mockDriver.EXPECT().Pop(s.worker.queue).Return(nil, errors.QueueDriverNoJobFound).Once()

// Start the worker
go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Give worker time to start
time.Sleep(100 * time.Millisecond)

// Call shutdown - it should wait for WaitGroup
err := s.worker.Shutdown()

s.NoError(err)
s.shutdownWorker(runErrChan)
s.True(s.worker.isShutdown.Load())
})

Expand Down Expand Up @@ -1038,16 +994,13 @@ func (s *WorkerTestSuite) TestShutdown() {
mockReservedJob.EXPECT().Delete().Return(nil).Once()

// Start the worker
go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Give worker time to start processing
time.Sleep(100 * time.Millisecond)

// Shutdown should wait for the job to complete
err := s.worker.Shutdown()
s.NoError(err)
s.shutdownWorker(runErrChan)
s.True(called)
})

Expand Down Expand Up @@ -1086,16 +1039,13 @@ func (s *WorkerTestSuite) TestShutdown() {
mockQuery.EXPECT().Insert(matchFailedJob("test-error")).Return(nil, nil).Once()

// Start the worker
go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Give worker time to process the failed job
time.Sleep(200 * time.Millisecond)

// Shutdown should wait for failed job logger goroutine
err := s.worker.Shutdown()
s.NoError(err)
s.shutdownWorker(runErrChan)
})

s.Run("shutdown when a task is running but fails finally", func() {
Expand Down Expand Up @@ -1136,16 +1086,13 @@ func (s *WorkerTestSuite) TestShutdown() {
mockQuery.EXPECT().Insert(matchFailedJob("test-error")).Return(nil, nil).Once()

// Start the worker
go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Give worker time to process the failed job
time.Sleep(200 * time.Millisecond)

// Shutdown should wait for failed job logger goroutine
err := s.worker.Shutdown()
s.NoError(err)
s.shutdownWorker(runErrChan)
})

s.Run("shutdown with multiple jobs", func() {
Expand Down Expand Up @@ -1180,19 +1127,39 @@ func (s *WorkerTestSuite) TestShutdown() {
s.mockDriver.EXPECT().Pop(s.worker.queue).Return(nil, errors.QueueDriverNoJobFound).Twice()

// Start the worker
go func() {
s.NoError(s.worker.run())
}()
runErrChan := s.startWorker()

// Give workers time to process
time.Sleep(200 * time.Millisecond)

// Shutdown should wait for all workers to complete
err := s.worker.Shutdown()
s.NoError(err)
s.shutdownWorker(runErrChan)
})
}

// startWorker runs the worker in the background and returns a channel that receives run's result.
// Asserting on that result from the subtest goroutine, instead of inside the background goroutine,
// keeps it from racing with testify resetting the suite's T once the subtest ends.
func (s *WorkerTestSuite) startWorker() <-chan error {
runErrChan := make(chan error, 1)
go func() {
runErrChan <- s.worker.run()
}()

return runErrChan
}

// shutdownWorker shuts the worker down and waits for the run started by startWorker to return.
func (s *WorkerTestSuite) shutdownWorker(runErrChan <-chan error) {
s.NoError(s.worker.Shutdown())
select {
case err := <-runErrChan:
s.NoError(err)
case <-time.After(5 * time.Second):
s.Fail("the worker did not stop after shutdown")
}
}

// matchFailedJob is a helper to match FailedJob with specific UUID
func matchFailedJob(uuid string) interface{} {
return mock.MatchedBy(func(job *models.FailedJob) bool {
Expand Down
Loading