Skip to content

Commit f3aa68f

Browse files
siherrmannsiherrmann
authored
fix wait timeout if job already finished (#16)
1 parent 18e606e commit f3aa68f

4 files changed

Lines changed: 34 additions & 64 deletions

File tree

‎example/exampleEasy.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package main
33
import (
44
"context"
55
"log"
6+
"time"
67

78
"github.com/siherrmann/queuer"
89
)
@@ -25,7 +26,7 @@ func ExampleEasy() {
2526
log.Fatalf("Error adding job: %v", err)
2627
}
2728

28-
job = q.WaitForJobFinished(job.RID)
29+
job = q.WaitForJobFinished(job.RID, 30*time.Second)
2930

3031
log.Printf("Job finished with status: %s", job.Status)
3132

‎example/exampleTx.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package main
33
import (
44
"context"
55
"log"
6+
"time"
67

78
"github.com/siherrmann/queuer"
89
)
@@ -47,7 +48,7 @@ func ExampleTx() {
4748
}
4849

4950
// Wait for job to finish for stopping the queuer
50-
job = q.WaitForJobFinished(job.RID)
51+
job = q.WaitForJobFinished(job.RID, 30*time.Second)
5152

5253
log.Printf("Job finished with status: %s", job.Status)
5354

‎queuerJob.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,13 @@ func (q *Queuer) WaitForJobAdded() *model.Job {
162162

163163
// WaitForJobFinished waits for a job to finish and returns the job.
164164
// It listens for job delete events and returns the job when it is finished.
165-
func (q *Queuer) WaitForJobFinished(jobRid uuid.UUID) *model.Job {
165+
// If timeout is reached before the job finishes, it returns nil.
166+
func (q *Queuer) WaitForJobFinished(jobRid uuid.UUID, timeout time.Duration) *model.Job {
167+
// First check if the job is already finished (in archive)
168+
if archivedJob, err := q.dbJob.SelectJobFromArchive(jobRid); err == nil && archivedJob != nil {
169+
return archivedJob
170+
}
171+
166172
jobFinished := make(chan *model.Job, 1)
167173
outerReady := make(chan struct{})
168174
ready := make(chan struct{})
@@ -177,10 +183,14 @@ func (q *Queuer) WaitForJobFinished(jobRid uuid.UUID) *model.Job {
177183

178184
<-outerReady
179185
<-ready
186+
187+
timeoutChan := time.After(timeout)
180188
for {
181189
select {
182190
case job := <-jobFinished:
183191
return job
192+
case <-timeoutChan:
193+
return nil
184194
case <-q.ctx.Done():
185195
return nil
186196
}

‎queuerJob_test.go‎

Lines changed: 19 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -116,23 +116,9 @@ func TestAddJobRunning(t *testing.T) {
116116
assert.NoError(t, err, "GetJob should not return an error")
117117
assert.NotNil(t, queuedJob, "GetJob should return the job that is currently running")
118118

119-
done := make(chan struct{})
120-
go func() {
121-
job = testQueuer.WaitForJobFinished(job.RID)
122-
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
123-
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
124-
close(done)
125-
}()
126-
127-
outerloop:
128-
for {
129-
select {
130-
case <-done:
131-
break outerloop
132-
case <-time.After(5 * time.Second):
133-
t.Fatal("WaitForJobFinished timed out waiting for job to finish")
134-
}
135-
}
119+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
120+
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
121+
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
136122

137123
jobNotExisting, err := testQueuer.GetJob(job.RID)
138124
assert.Error(t, err, "GetJob should return an error for ended job")
@@ -161,23 +147,9 @@ func TestAddJobRunning(t *testing.T) {
161147
require.NotNil(t, queuedJob, "GetJob should return the job that is currently running")
162148
assert.Equal(t, model.JobStatusScheduled, queuedJob.Status, "Job should be in Running status")
163149

164-
done := make(chan struct{})
165-
go func() {
166-
job = testQueuer.WaitForJobFinished(job.RID)
167-
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
168-
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
169-
close(done)
170-
}()
171-
172-
outerloop:
173-
for {
174-
select {
175-
case <-done:
176-
break outerloop
177-
case <-time.After(5 * time.Second):
178-
t.Fatal("WaitForJobFinished timed out waiting for job to finish")
179-
}
180-
}
150+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
151+
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
152+
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
181153

182154
// Check if the job is archived
183155
jobNotExisting, err := testQueuer.GetJob(job.RID)
@@ -204,23 +176,9 @@ func TestAddJobRunning(t *testing.T) {
204176
require.NotNil(t, job, "GetJob should return the job that is currently running")
205177
assert.Equal(t, model.JobStatusScheduled, job.Status, "Job should be in Scheduled status")
206178

207-
done := make(chan struct{})
208-
go func() {
209-
job = testQueuer.WaitForJobFinished(job.RID)
210-
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
211-
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
212-
close(done)
213-
}()
214-
215-
outerloop:
216-
for {
217-
select {
218-
case <-done:
219-
break outerloop
220-
case <-time.After(5 * time.Second):
221-
t.Fatal("WaitForJobFinished timed out waiting for job to finish")
222-
}
223-
}
179+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
180+
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
181+
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
224182

225183
// Check if the job is archived
226184
jobNotExisting, err := testQueuer.GetJob(job.RID)
@@ -514,12 +472,12 @@ func TestAddJobWithScheduleOptionsRunning(t *testing.T) {
514472

515473
log.Printf("Second job started: %v", secondJob)
516474

517-
secondJob = testQueuer.WaitForJobFinished(secondJob.RID)
475+
secondJob = testQueuer.WaitForJobFinished(secondJob.RID, 10*time.Second)
518476
assert.Equal(t, model.JobStatusSucceeded, secondJob.Status, "Second job should have status Succeeded")
519477
close(secondJobFinished)
520478
}()
521479

522-
job = testQueuer.WaitForJobFinished(job.RID)
480+
job = testQueuer.WaitForJobFinished(job.RID, 10*time.Second)
523481
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
524482
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
525483

@@ -723,7 +681,7 @@ func TestWaitForJobFinished(t *testing.T) {
723681
job, err := testQueuer.AddJob(TaskMock, 1, "2")
724682
assert.NoError(t, err, "AddJob should not return an error on success")
725683

726-
job = testQueuer.WaitForJobFinished(job.RID)
684+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
727685
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
728686
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
729687

@@ -734,15 +692,15 @@ func TestWaitForJobFinished(t *testing.T) {
734692
})
735693

736694
t.Run("Successfully cancel context while waiting for job", func(t *testing.T) {
737-
job, err := testQueuer.AddJob(TaskMock, 1, "2")
695+
job, err := testQueuer.AddJob(TaskMock, 10, "2")
738696
assert.NoError(t, err, "AddJob should not return an error on success")
739697

740698
go func() {
741699
time.Sleep(500 * time.Millisecond)
742700
cancel()
743701
}()
744702

745-
job = testQueuer.WaitForJobFinished(job.RID)
703+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
746704
assert.Nil(t, job, "WaitForJobFinished should return nil when context is cancelled")
747705
})
748706

@@ -963,12 +921,12 @@ func TestGetJobsEndedBySearch(t *testing.T) {
963921
require.NotNil(t, job2, "Job2 should not be nil")
964922

965923
// Wait for jobs to finish
966-
finishedJob1 := testQueuer.WaitForJobFinished(job1.RID)
967-
assert.NotNil(t, finishedJob1, "Job1 should finish")
924+
finishedJob1 := testQueuer.WaitForJobFinished(job1.RID, 30*time.Second)
925+
require.NotNil(t, finishedJob1, "Job1 should finish within timeout")
968926
assert.Contains(t, []string{model.JobStatusSucceeded, model.JobStatusFailed}, finishedJob1.Status, "Job1 should have finished")
969927

970-
finishedJob2 := testQueuer.WaitForJobFinished(job2.RID)
971-
assert.NotNil(t, finishedJob2, "Job2 should finish")
928+
finishedJob2 := testQueuer.WaitForJobFinished(job2.RID, 30*time.Second)
929+
require.NotNil(t, finishedJob2, "Job2 should finish within timeout")
972930
assert.Contains(t, []string{model.JobStatusSucceeded, model.JobStatusFailed}, finishedJob2.Status, "Job2 should have finished")
973931

974932
// Search for ended jobs by task name
@@ -1004,7 +962,7 @@ func TestDeleteJob(t *testing.T) {
1004962
job, err := testQueuer.AddJob(TaskMock, 1, "2")
1005963
assert.NoError(t, err, "AddJob should not return an error on success")
1006964

1007-
job = testQueuer.WaitForJobFinished(job.RID)
965+
job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second)
1008966
assert.NotNil(t, job, "WaitForJobFinished should return the finished job")
1009967
assert.Equal(t, model.JobStatusSucceeded, job.Status, "WaitForJobFinished should return job with status Succeeded")
1010968

0 commit comments

Comments
 (0)