Fix Delete race and prioritize write ops over tirgger execution (#31)

* Fix Delete race and prioritize write ops over tirgger execution

Fixes a race condition whereby a Delete operation on a job which was mid
execution on the same scheduler instance, would not see that job as not
deleted from the in-memory.

Adds fix for this by tracking names of active jobs, and backing out of
an execution if it is no longer in that cache (deleted).

Using RWMutex, priorities Delete/Write operations over job trigger
execution when on local node.

Signed-off-by: joshvanl <[email protected]>

* Update dapr/kit go.mod

Signed-off-by: joshvanl <[email protected]>

---------

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

Makefile

@@ -72,4 +72,4 @@ GEN_PROTOS:=$(foreach ITEM,$(PROTOS),gen-proto-$(ITEM))
 gen-proto: check-proto-version $(GEN_PROTOS) modtidy
 
 test:
-	go test -timeout 300s --race ./...
+	go test -count 1 -timeout 300s --race ./...

api.go

@@ -95,11 +95,17 @@ func (c *cron) Delete(ctx context.Context, name string) error {
 		return err
 	}
 
-	if _, err := c.client.Delete(ctx, c.key.JobKey(name)); err != nil {
+	jobKey := c.key.JobKey(name)
+
+	c.queueLock.Lock(jobKey)
+	defer c.queueLock.Unlock(jobKey)
+
+	if _, err := c.client.Delete(ctx, jobKey); err != nil {
 		return err
 	}
 
-	return nil
+	c.queueCache.Delete(jobKey)
+	return c.queue.Dequeue(jobKey)
 }
 
 // validateName validates the name of a job.

cron.go

@@ -107,7 +107,8 @@ type cron struct {
 	wg      sync.WaitGroup
 	// queueLock prevents an informed schedule from overwriting a job as it is
 	// being triggered, i.e. prevent a PUT and mid-trigger race condition.
-	queueLock concurrency.MutexMap[string]
+	queueLock  concurrency.MutexMap[string]
+	queueCache *sync.Map
 }
 
 // New creates a new cron instance.
@@ -179,6 +180,7 @@ func New(opts Options) (Interface, error) {
 		closeCh:              make(chan struct{}),
 		errCh:                make(chan error),
 		queueLock:            concurrency.NewMutexMap[string](),
+		queueCache:           new(sync.Map),
 	}, nil
 }
 
@@ -191,19 +193,21 @@ func (c *cron) Run(ctx context.Context) error {
 
 	c.queue = queue.NewProcessor[string, *counter.Counter](
 		func(counter *counter.Counter) {
-			c.queueLock.Lock(counter.Key())
-			if ctx.Err() != nil {
-				c.queueLock.Unlock(counter.Key())
+			c.queueLock.RLock(counter.Key())
+			_, ok := c.queueCache.Load(counter.Key())
+			if !ok || ctx.Err() != nil {
+				c.queueLock.DeleteRUnlock(counter.Key())
 				return
 			}
 
 			c.wg.Add(1)
 			go func() {
 				defer c.wg.Done()
 				if c.handleTrigger(ctx, counter) {
-					c.queueLock.Unlock(counter.Key())
+					c.queueLock.RUnlock(counter.Key())
 				} else {
-					c.queueLock.DeleteUnlock(counter.Key())
+					c.queueCache.Delete(counter.Key())
+					c.queueLock.DeleteRUnlock(counter.Key())
 				}
 			}()
 		},
@@ -313,6 +317,7 @@ func (c *cron) handleInformerEvent(ctx context.Context, e *informer.Event) error
 	}
 
 	defer c.queueLock.DeleteUnlock(string(e.Key))
+	c.queueCache.Delete(string(e.Key))
 	return c.queue.Dequeue(string(e.Key))
 }
 
@@ -342,5 +347,6 @@ func (c *cron) schedule(ctx context.Context, name string, job *api.JobStored) er
 	default:
 	}
 
+	c.queueCache.Store(counter.Key(), nil)
 	return c.queue.Enqueue(counter)
 }

cron_test.go

@@ -170,7 +170,7 @@ func Test_repeat(t *testing.T) {
 	helper := testCron(t, 1)
 
 	job := &api.Job{
-		Schedule: ptr.Of("@every 1s"),
+		Schedule: ptr.Of("@every 10ms"),
 		Repeats:  ptr.Of(uint32(3)),
 	}
 
@@ -523,7 +523,7 @@ func Test_parallel(t *testing.T) {
 			var done atomic.Int32
 			helper := testCronWithOptions(t, testCronOptions{
 				total: total,
-				triggerFn: func() {
+				triggerFn: func(*api.TriggerRequest) {
 					waiting.Add(1)
 					<-releaseCh
 					done.Add(1)
@@ -547,11 +547,66 @@ func Test_parallel(t *testing.T) {
 	}
 }
 
+func Test_DeleteRace(t *testing.T) {
+	t.Parallel()
+
+	triggered := make([]atomic.Int64, 20)
+	helper := testCronWithOptions(t, testCronOptions{
+		total: 1,
+		triggerFn: func(req *api.TriggerRequest) {
+			i, err := strconv.Atoi(req.Name)
+			require.NoError(t, err)
+			triggered[i].Add(1)
+		},
+	})
+
+	jobNames := make([]string, 20)
+	for i := range jobNames {
+		jobNames[i] = strconv.Itoa(i)
+		require.NoError(t, helper.cron.Add(helper.ctx, jobNames[i], &api.Job{
+			Schedule: ptr.Of("@every 1s"),
+		}))
+	}
+
+	cron := helper.cron.(*cron)
+
+	for i, name := range jobNames {
+		i := i
+		name := name
+		keyName := "abc/jobs/" + name
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			assert.EventuallyWithT(t, func(c *assert.CollectT) {
+				_, ok := cron.queueCache.Load(keyName)
+				assert.True(c, ok)
+			}, 5*time.Second, time.Millisecond)
+
+			assert.EventuallyWithT(t, func(c *assert.CollectT) {
+				assert.GreaterOrEqual(c, triggered[i].Load(), int64(1))
+			}, 5*time.Second, time.Millisecond)
+
+			_, ok := cron.queueCache.Load(keyName)
+			assert.True(t, ok)
+			require.NoError(t, helper.cron.Delete(helper.ctx, name))
+
+			_, ok = cron.queueCache.Load(keyName)
+			assert.False(t, ok)
+
+			currentTriggered := triggered[i].Load()
+			time.Sleep(time.Second * 2)
+			assert.Equal(t, currentTriggered, triggered[i].Load())
+			_, ok = cron.queueCache.Load(keyName)
+			assert.False(t, ok)
+		})
+	}
+}
+
 type testCronOptions struct {
 	total     uint32
 	returnOk  *atomic.Bool
 	gotCh     chan *api.TriggerRequest
-	triggerFn func()
+	triggerFn func(*api.TriggerRequest)
 }
 
 type helper struct {
@@ -592,7 +647,7 @@ func testCronWithOptions(t *testing.T, opts testCronOptions) *helper {
 					opts.gotCh <- req
 				}
 				if opts.triggerFn != nil {
-					opts.triggerFn()
+					opts.triggerFn(req)
 				}
 				return ok
 			},

go.mod

@@ -3,7 +3,7 @@ module github.com/diagridio/go-etcd-cron
 go 1.22.5
 
 require (
-	github.com/dapr/kit v0.13.1-0.20240722163453-58c6d9df14d3
+	github.com/dapr/kit v0.13.1-0.20240724000121-26b564d9d0f5
 	github.com/go-logr/logr v1.3.0
 	github.com/go-logr/zapr v1.3.0
 	github.com/stretchr/testify v1.9.0

go.sum

@@ -33,8 +33,8 @@ github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmf
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
-github.com/dapr/kit v0.13.1-0.20240722163453-58c6d9df14d3 h1:+HZd67sGtxQq7UoEXpby0x0pE0XcZwTY03UuZc48P/M=
-github.com/dapr/kit v0.13.1-0.20240722163453-58c6d9df14d3/go.mod h1:Hz1W2LmWfA4UX/12MdA+brsf+np6f/1dJt6C6F63cjI=
+github.com/dapr/kit v0.13.1-0.20240724000121-26b564d9d0f5 h1:FQKdGOG6Zi3gBhtnxPQmrd8QFLs8e6JTGkts+aaXba0=
+github.com/dapr/kit v0.13.1-0.20240724000121-26b564d9d0f5/go.mod h1:Hz1W2LmWfA4UX/12MdA+brsf+np6f/1dJt6C6F63cjI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=