adding basic gocron-gorm-lock functionality.

README.md

@@ -1,2 +1,58 @@
 # gocron-gorm-lock
 A gocron locker implementation using gorm
+
+## install
+
+```
+go get github.com/go-co-op/gocron-gorm-lock
+```
+
+## usage
+
+Here is an example usage that would be deployed in multiple instances
+
+```go
+package main
+
+import (
+	"fmt"
+
+	"github.com/go-co-op/gocron"
+	gormlock "github.com/go-co-op/gocron-gorm-lock"
+	"gorm.io/gorm"
+	"time"
+)
+
+func main() {
+	var db * gorm.DB // gorm db connection
+	var worker string // name of this instance to be used to know which instance run the job
+	db.AutoMigrate(&CronJobLock{}) // We need the table to store the job execution
+	locker, err := gormlock.NewGormLocker(db, worker)
+	if err != nil {
+		// handle the error
+	}
+
+	s := gocron.NewScheduler(time.UTC)
+	s.WithDistributedLocker(locker)
+
+	_, err = s.Every("1s").Name("unique_name").Do(func() {
+		// task to do
+		fmt.Println("call 1s")
+	})
+	if err != nil {
+		// handle the error
+	}
+
+	s.StartBlocking()
+}
+```
+
+## Prerequisites
+
+- The table cron_job_locks needs to exists in the database. This can be achieved, as an example, using gorm automigrate functionality `db.Automigrate(&CronJobLock{})`
+- In order to uniquely identify the job, the locker uses the unique combination of the job name + timestamp (by default with precision to miliseconds).
+
+## FAQ
+
+- The locker uses the unique combination of the job name + timestamp with miliseconds precision, how can I change that:
+- It's possible to set how to create the job identifier, here is an example to set an hour precision: `locker, err := gormlock.NewGormLocker(db, "local", WithJobIdentifier(func(ctx context.Context, key string) string { return time.Now().Truncate(60 * time.Minute).Format("2006-01-02 15:04:05.000")}))`

go.mod

@@ -1,3 +1,59 @@
 module github.com/go-co-op/gocron-gorm-lock
 
 go 1.20
+
+require (
+	github.com/go-co-op/gocron v1.31.0
+	github.com/stretchr/testify v1.8.4
+	github.com/testcontainers/testcontainers-go v0.22.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.22.0
+	gorm.io/driver/postgres v1.5.2
+	gorm.io/gorm v1.25.2
+)
+
+require (
+	dario.cat/mergo v1.0.0 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
+	github.com/Microsoft/go-winio v0.6.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.0 // indirect
+	github.com/containerd/containerd v1.7.3 // indirect
+	github.com/cpuguy83/dockercfg v0.3.1 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/docker/distribution v2.8.2+incompatible // indirect
+	github.com/docker/docker v24.0.5+incompatible // indirect
+	github.com/docker/go-connections v0.4.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgx/v5 v5.3.1 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/klauspost/compress v1.16.0 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/moby/patternmatcher v0.5.0 // indirect
+	github.com/moby/sys/sequential v0.5.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0-rc4 // indirect
+	github.com/opencontainers/runc v1.1.5 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
+	github.com/sirupsen/logrus v1.9.0 // indirect
+	go.uber.org/atomic v1.9.0 // indirect
+	golang.org/x/crypto v0.8.0 // indirect
+	golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect
+	golang.org/x/mod v0.9.0 // indirect
+	golang.org/x/net v0.9.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/tools v0.7.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect
+	google.golang.org/grpc v1.57.0 // indirect
+	google.golang.org/protobuf v1.30.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

gormlock.go

@@ -0,0 +1,62 @@
+package gormlock
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-co-op/gocron"
+	"gorm.io/gorm"
+)
+
+var defaultPrecision = time.Millisecond
+
+func NewGormLocker(db *gorm.DB, worker string, options ...LockOption) (gocron.Locker, error) {
+	gl := &gormLocker{db: db, worker: worker}
+	for _, option := range options {
+		option(gl)
+	}
+	return gl, nil
+}
+
+var _ gocron.Locker = (*gormLocker)(nil)
+
+type gormLocker struct {
+	db            *gorm.DB
+	worker        string
+	jobIdentifier func(ctx context.Context, key string) string
+}
+
+func (g *gormLocker) Lock(ctx context.Context, key string) (gocron.Lock, error) {
+	ji := g.getJobIdentifier(ctx, key)
+
+	// I would like that people can "pass" their own implementation,
+	cjb := &CronJobLock{
+		JobName:       key,
+		JobIdentifier: ji,
+		Worker:        g.worker,
+		Status:        "RUNNING",
+	}
+	tx := g.db.Create(cjb)
+	if tx.Error != nil {
+		return nil, tx.Error
+	}
+	return &gormLock{db: g.db, id: cjb.GetID()}, nil
+}
+
+func (g *gormLocker) getJobIdentifier(ctx context.Context, key string) string {
+	if g.jobIdentifier == nil {
+		return time.Now().Truncate(defaultPrecision).Format("2006-01-02 15:04:05.000")
+	}
+	return g.jobIdentifier(ctx, key)
+}
+
+var _ gocron.Lock = (*gormLock)(nil)
+
+type gormLock struct {
+	db *gorm.DB
+	id int
+}
+
+func (g *gormLock) Unlock(_ context.Context) error {
+	return g.db.Model(&CronJobLock{ID: g.id}).Updates(&CronJobLock{Status: "FINISHED"}).Error
+}

gormlock_test.go

@@ -0,0 +1,208 @@
+package gormlock
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/go-co-op/gocron"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go"
+	testcontainerspostgres "github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestEnableDistributedLocking(t *testing.T) {
+	ctx := context.Background()
+	postgresContainer, err := testcontainerspostgres.RunContainer(ctx,
+		testcontainers.WithWaitStrategy(wait.ForLog("database system is ready to accept connections").
+			WithOccurrence(2).WithStartupTimeout(5*time.Second)))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := postgresContainer.Terminate(ctx); err != nil {
+			t.Fatalf("failed to terminate container: %s", err)
+		}
+	})
+
+	connStr, err := postgresContainer.ConnectionString(ctx, "sslmode=disable", "application_name=test")
+	assert.NoError(t, err)
+
+	db, err := gorm.Open(postgres.Open(connStr), &gorm.Config{})
+	require.NoError(t, err)
+
+	err = db.AutoMigrate(&CronJobLock{})
+	require.NoError(t, err)
+
+	resultChan := make(chan int, 10)
+	f := func(schedulerInstance int) {
+		resultChan <- schedulerInstance
+	}
+
+	s1 := gocron.NewScheduler(time.UTC)
+	l1, err := NewGormLocker(db, "s1")
+	require.NoError(t, err)
+	s1.WithDistributedLocker(l1)
+	_, err = s1.Every("500ms").Do(f, 1)
+	require.NoError(t, err)
+
+	s2 := gocron.NewScheduler(time.UTC)
+	l2, err := NewGormLocker(db, "s2")
+	require.NoError(t, err)
+	s2.WithDistributedLocker(l2)
+	_, err = s2.Every("500ms").Do(f, 2)
+	require.NoError(t, err)
+
+	s3 := gocron.NewScheduler(time.UTC)
+	l3, err := NewGormLocker(db, "s3")
+	require.NoError(t, err)
+	s3.WithDistributedLocker(l3)
+	_, err = s3.Every("500ms").Do(f, 3)
+	require.NoError(t, err)
+
+	s1.StartAsync()
+	s2.StartAsync()
+	s3.StartAsync()
+
+	time.Sleep(1700 * time.Millisecond)
+
+	s1.Stop()
+	s2.Stop()
+	s3.Stop()
+	close(resultChan)
+
+	var results []int
+	for r := range resultChan {
+		results = append(results, r)
+	}
+	assert.Len(t, results, 4)
+	var allCronJobs []*CronJobLock
+	db.Find(&allCronJobs)
+	assert.Equal(t, len(results), len(allCronJobs))
+}
+
+func TestEnableDistributedLocking_DifferentJob(t *testing.T) {
+	ctx := context.Background()
+	postgresContainer, err := testcontainerspostgres.RunContainer(ctx,
+		testcontainers.WithWaitStrategy(wait.ForLog("database system is ready to accept connections").
+			WithOccurrence(2).WithStartupTimeout(5*time.Second)))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := postgresContainer.Terminate(ctx); err != nil {
+			t.Fatalf("failed to terminate container: %s", err)
+		}
+	})
+
+	connStr, err := postgresContainer.ConnectionString(ctx, "sslmode=disable", "application_name=test")
+	assert.NoError(t, err)
+
+	db, err := gorm.Open(postgres.Open(connStr), &gorm.Config{})
+	require.NoError(t, err)
+
+	err = db.AutoMigrate(&CronJobLock{})
+	require.NoError(t, err)
+
+	resultChan := make(chan int, 10)
+	f := func(schedulerInstance int) {
+		resultChan <- schedulerInstance
+	}
+
+	result2Chan := make(chan int, 10)
+	f2 := func(schedulerInstance int) {
+		result2Chan <- schedulerInstance
+	}
+
+	s1 := gocron.NewScheduler(time.UTC)
+	l1, err := NewGormLocker(db, "s1")
+	require.NoError(t, err)
+	s1.WithDistributedLocker(l1)
+	_, err = s1.Every("500ms").Name("f").Do(f, 1)
+	require.NoError(t, err)
+	_, err = s1.Every("500ms").Name("f2").Do(f2, 1)
+	require.NoError(t, err)
+
+	s2 := gocron.NewScheduler(time.UTC)
+	l2, err := NewGormLocker(db, "s2")
+	require.NoError(t, err)
+	s2.WithDistributedLocker(l2)
+	_, err = s2.Every("500ms").Name("f").Do(f, 2)
+	require.NoError(t, err)
+	_, err = s2.Every("500ms").Name("f2").Do(f2, 2)
+	require.NoError(t, err)
+
+	s3 := gocron.NewScheduler(time.UTC)
+	l3, err := NewGormLocker(db, "s3")
+	require.NoError(t, err)
+	s3.WithDistributedLocker(l3)
+	_, err = s3.Every("500ms").Name("f").Do(f, 3)
+	require.NoError(t, err)
+	_, err = s3.Every("500ms").Name("f2").Do(f2, 3)
+	require.NoError(t, err)
+
+	s1.StartAsync()
+	s2.StartAsync()
+	s3.StartAsync()
+
+	time.Sleep(1700 * time.Millisecond)
+
+	s1.Stop()
+	s2.Stop()
+	s3.Stop()
+	close(resultChan)
+	close(result2Chan)
+
+	var results []int
+	for r := range resultChan {
+		results = append(results, r)
+	}
+	assert.Len(t, results, 4, "f is expected 4 times")
+	var results2 []int
+	for r := range result2Chan {
+		results2 = append(results2, r)
+	}
+	assert.Len(t, results2, 4, "f2 is expected 4 times")
+	var allCronJobs []*CronJobLock
+	db.Find(&allCronJobs)
+	assert.Equal(t, len(results)+len(results2), len(allCronJobs))
+}
+
+func TestJobReturningExceptionWhenUnique(t *testing.T) {
+	ctx := context.Background()
+	postgresContainer, err := testcontainerspostgres.RunContainer(ctx,
+		testcontainers.WithWaitStrategy(wait.ForLog("database system is ready to accept connections").
+			WithOccurrence(2).WithStartupTimeout(5*time.Second)))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := postgresContainer.Terminate(ctx); err != nil {
+			t.Fatalf("failed to terminate container: %s", err)
+		}
+	})
+
+	connStr, err := postgresContainer.ConnectionString(ctx, "sslmode=disable", "application_name=test")
+	assert.NoError(t, err)
+
+	db, err := gorm.Open(postgres.Open(connStr), &gorm.Config{})
+	require.NoError(t, err)
+
+	err = db.AutoMigrate(&CronJobLock{})
+	require.NoError(t, err)
+
+	// creating a entry to force the unique identifier error
+	cjb := &CronJobLock{
+		JobName:       "job",
+		JobIdentifier: time.Now().Truncate(60 * time.Minute).Format("2006-01-02 15:04:05.000"),
+		Worker:        "local",
+		Status:        "RUNNING",
+	}
+	require.NoError(t, db.Create(cjb).Error)
+
+	l, _ := NewGormLocker(db, "local", WithJobIdentifier(func(ctx context.Context, key string) string {
+		return time.Now().Truncate(60 * time.Minute).Format("2006-01-02 15:04:05.000")
+	}))
+	_, lerr := l.Lock(ctx, "job")
+	if assert.Error(t, lerr) {
+		assert.ErrorContains(t, lerr, "violates unique constraint")
+	}
+}