Repository: go-co-op/gocron (go)

Issue:
[FEATURE] - Allow Disabling Global Locker At Job Level
### Is your feature request related to a problem? Please describe

There are occasions that we want our general schedulers to have a locking mechanism and instances of scheduler run only one job across all instances for all the defined job except some of them. In other words, let a specific job to be scheduled by all the scheduler across different instances.

### Describe the solution you'd like

#### 1. Edit WithDistributedJobLocker

https://github.com/go-co-op/gocron/blob/c180381e94fcc7061c643926e9af365d9205e55c/job.go#L552-L560

```go
func WithDistributedJobLocker(locker Locker) JobOption {
	return func(j *internalJob, _ time.Time) error {
		if locker == nil {
			j.disableGlobalLocker = true // new struct property
		} else {
		       j.locker = locker
                }
		return nil
	}
}
```

#### 2. Update `executor.runJob` to respect the above

https://github.com/go-co-op/gocron/blob/c180381e94fcc7061c643926e9af365d9205e55c/executor.go#L373-L391

The last else would be:

```go
...
else if e.locker != nil && !j.disableGlobalLocker {
        ...
}
```

### Describe alternatives you've considered

The only way I have came to is to implement a custom locker on my side and drop using the locking mechanism of gocron as these methods and functions are not part of the package's public API.

### Additional context

I'd be happy to open a PR and address this enhancement.
This feature would allow to run some of our jobs in batch mode by all the instance (and let the job itself decide how they want to divide and conquer if necessary) and some of our jobs only to be run by one instance at each point in time.


Pull request:
feat: allow disabling global distributed locker per job
### What does this do?
These changes allow us to disable the globally set distributed locker for a specific job, allowing those scheduled jobs to potentially run multiple times at once by different schedulers.

### Which issue(s) does this PR fix/relate to?
Resolves #808 

### List any changes that modify/break current functionality
A new job option has been introduced to remove any potential breaking changes. Suppose we use the existing `WithDistributedJobLocker` method and treat its `nil` argument as an act of disabling the global locker for the job. In that case, the method won't return an error anymore; therefore, `ErrWithDistributedJobLockerNil` will be unused.

### Have you included tests for your changes?
Yes, a new test case has been added to the distributed locker table test to check whether disabling the job locker while global locker is set works as expected.

### Did you document any new/modified functionality?

- [x] Updated `example_test.go` (didn't need to)
- [x] Updated `README.md` (didn't need to)
- [x] Added method documentation block
- [x] Updated WithDistributedLocker method documentation to mention this feature

### Notes
- A new job option instead of tweaking the existing job option to avoid potential breaking changes
- Tweaking distributed locker table test format to accommodate the new option for disabling it per job

Patch:
diff --git a/executor.go b/executor.go
index 5f0995a..439cacb 100644
--- a/executor.go
+++ b/executor.go
@@ -370,7 +370,7 @@ func (e *executor) runJob(j internalJob, jIn jobIn) {
 			e.incrementJobCounter(j, Skip)
 			return
 		}
-	} else if j.locker != nil {
+	} else if !j.disabledLocker && j.locker != nil {
 		lock, err := j.locker.Lock(j.ctx, j.name)
 		if err != nil {
 			_ = callJobFuncWithParams(j.afterLockError, j.id, j.name, err)
@@ -379,7 +379,7 @@ func (e *executor) runJob(j internalJob, jIn jobIn) {
 			return
 		}
 		defer func() { _ = lock.Unlock(j.ctx) }()
-	} else if e.locker != nil {
+	} else if !j.disabledLocker && e.locker != nil {
 		lock, err := e.locker.Lock(j.ctx, j.name)
 		if err != nil {
 			_ = callJobFuncWithParams(j.afterLockError, j.id, j.name, err)
diff --git a/job.go b/job.go
index 2ea8e91..4d868e6 100644
--- a/job.go
+++ b/job.go
@@ -45,6 +45,7 @@ type internalJob struct {
 	afterJobRunsWithError func(jobID uuid.UUID, jobName string, err error)
 	afterJobRunsWithPanic func(jobID uuid.UUID, jobName string, recoverData any)
 	afterLockError        func(jobID uuid.UUID, jobName string, err error)
+	disabledLocker        bool
 
 	locker Locker
 }
@@ -556,6 +557,16 @@ func WithDistributedJobLocker(locker Locker) JobOption {
 	}
 }
 
+// WithDisabledDistributedJobLocker disables the distributed job locker.
+// This is useful when a global distributed locker has been set on the scheduler
+// level using WithDistributedLocker and need to be disabled for specific jobs.
+func WithDisabledDistributedJobLocker(disabled bool) JobOption {
+	return func(j *internalJob, _ time.Time) error {
+		j.disabledLocker = disabled
+		return nil
+	}
+}
+
 // WithEventListeners sets the event listeners that should be
 // run for the job.
 func WithEventListeners(eventListeners ...EventListener) JobOption {
diff --git a/scheduler.go b/scheduler.go
index 9827f0c..825323e 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -818,6 +818,8 @@ func WithDistributedElector(elector Elector) SchedulerOption {
 // WithDistributedLocker sets the locker to be used by multiple
 // Scheduler instances to ensure that only one instance of each
 // job is run.
+// To disable this global locker for specific jobs, see
+// WithDisabledDistributedJobLocker.
 func WithDistributedLocker(locker Locker) SchedulerOption {
 	return func(s *scheduler) error {
 		if locker == nil {


Test patch:
diff --git a/scheduler_test.go b/scheduler_test.go
index d335c4a..8d46d31 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -1452,6 +1452,7 @@ func TestScheduler_WithDistributed(t *testing.T) {
 	tests := []struct {
 		name          string
 		count         int
+		runCount      int
 		schedulerOpts []SchedulerOption
 		jobOpts       []JobOption
 		assertions    func(*testing.T)
@@ -1459,6 +1460,7 @@ func TestScheduler_WithDistributed(t *testing.T) {
 		{
 			"3 schedulers with elector",
 			3,
+			1,
 			[]SchedulerOption{
 				WithDistributedElector(&testElector{notLeader: notLeader}),
 			},
@@ -1482,6 +1484,7 @@ func TestScheduler_WithDistributed(t *testing.T) {
 		{
 			"3 schedulers with locker",
 			3,
+			1,
 			[]SchedulerOption{
 				WithDistributedLocker(&testLocker{notLocked: notLocked}),
 			},
@@ -1499,11 +1502,14 @@ func TestScheduler_WithDistributed(t *testing.T) {
 					default:
 					}
 				}
+
+				assert.Equal(t, 2, notLockedCount)
 			},
 		},
 		{
 			"3 schedulers and job with Distributed locker",
 			3,
+			1,
 			nil,
 			[]JobOption{
 				WithDistributedJobLocker(&testLocker{notLocked: notLocked}),
@@ -1521,6 +1527,35 @@ func TestScheduler_WithDistributed(t *testing.T) {
 					default:
 					}
 				}
+
+				assert.Equal(t, 2, notLockedCount)
+			},
+		},
+		{
+			"3 schedulers and job with disabled Distributed locker",
+			3,
+			3,
+			[]SchedulerOption{
+				WithDistributedLocker(&testLocker{notLocked: notLocked}),
+			},
+			[]JobOption{
+				WithDisabledDistributedJobLocker(true),
+			},
+			func(_ *testing.T) {
+				timeout := time.Now().Add(1 * time.Second)
+				var notLockedCount int
+				for {
+					if time.Now().After(timeout) {
+						break
+					}
+					select {
+					case <-notLocked:
+						notLockedCount++
+					default:
+					}
+				}
+
+				assert.Equal(t, 0, notLockedCount)
 			},
 		},
 	}
@@ -1531,6 +1566,11 @@ func TestScheduler_WithDistributed(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
 			schedulersDone := make(chan struct{}, tt.count)
 
+			var (
+				runCount  int
+				doneCount int
+			)
+
 			for i := tt.count; i > 0; i-- {
 				s := newTestScheduler(t,
 					tt.schedulerOpts...,
@@ -1539,6 +1579,7 @@ func TestScheduler_WithDistributed(t *testing.T) {
 					WithStartAt(
 						WithStartImmediately(),
 					),
+					WithLimitedRuns(1),
 				}
 				jobOpts = append(jobOpts, tt.jobOpts...)
 
@@ -1565,31 +1606,39 @@ func TestScheduler_WithDistributed(t *testing.T) {
 				}()
 			}
 
-			var runCount int
-			select {
-			case <-jobsRan:
-				cancel()
-				runCount++
-			case <-time.After(time.Second):
-				cancel()
-				t.Error("timed out waiting for job to run")
+		RunCountLoop:
+			for {
+				select {
+				case <-jobsRan:
+					runCount++
+					if runCount >= tt.runCount {
+						break RunCountLoop
+					}
+				case <-time.After(time.Second):
+					t.Error("timed out waiting for job to run")
+					break RunCountLoop
+				}
 			}
 
-			var doneCount int
-			timeout := time.Now().Add(3 * time.Second)
-			for doneCount < tt.count && time.Now().After(timeout) {
+			cancel()
+			assert.Equal(t, tt.runCount, runCount)
+
+		DoneCountLoop:
+			for {
 				select {
 				case <-schedulersDone:
 					doneCount++
-				default:
+					if doneCount >= tt.count {
+						break DoneCountLoop
+					}
+				case <-time.After(3 * time.Second):
+					t.Error("timed out waiting for schedulers to shutdown")
+					break DoneCountLoop
 				}
 			}
-			close(jobsRan)
-			for range jobsRan {
-				runCount++
-			}
 
-			assert.Equal(t, 1, runCount)
+			assert.Equal(t, tt.count, doneCount)
+
 			time.Sleep(time.Second)
 			tt.assertions(t)
 		})