fix: prevent duplicate cron entries from concurrent SyncCanaryJob calls (#2984)

* test(sync): add concurrent SyncCanaryJob reschedule race test

Spawns 10 goroutines calling SyncCanaryJob concurrently with
alternating specs (forcing a changed detection). Without the
per-canary-ID lock, the race between Unschedule and newCanaryJob
produces multiple orphaned cron entries that survive cleanup sweeps.

* fix(sync): prevent duplicate cron entries from concurrent SyncCanaryJob calls

Concurrent calls to SyncCanaryJob (from the controller reconcile and the
periodic SyncCanaryJobs job) can race: the first call Unschedule's and
deletes the canaryJobs map entry, the second sees nil and creates a
cron entry, then the first also creates a second cron entry. The
orphaned entry survives all cleanup sweeps (SetDifference sees the
same UID in both lists) and fires on every schedule tick, doubling
check_statuses inserts and inflating aggregated fail counts.

Serialize SyncCanaryJob per canary ID with a lazily-initialized mutex
from a canarySyncLocks sync.Map to close the race window.

Author: adityathebe

go.mod

@@ -44,6 +44,7 @@ require (
 	github.com/labstack/echo/v4 v4.15.1
 	github.com/liamylian/jsontime/v2 v2.0.0
 	github.com/lib/pq v1.12.3
+	github.com/mdelapenya/tlscert v0.2.0
 	github.com/microsoft/azure-devops-go-api/azuredevops/v7 v7.1.0
 	github.com/microsoft/go-mssqldb v1.9.8
 	github.com/nsf/jsondiff v0.0.0-20230430225905-43f6cf3098c1
@@ -66,6 +67,7 @@ require (
 	github.com/sevennt/echo-pprof v0.1.1-0.20220616082843-66a461746b5f
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.10
+	github.com/testcontainers/testcontainers-go v0.43.0
 	github.com/testcontainers/testcontainers-go/modules/redis v0.43.0
 	github.com/timberio/go-datemath v0.1.0
 	go.mongodb.org/mongo-driver v1.17.9
@@ -339,7 +341,6 @@ require (
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-localereader v0.0.1 // indirect
 	github.com/mattn/go-runewidth v0.0.21 // indirect
-	github.com/mdelapenya/tlscert v0.2.0 // indirect
 	github.com/mitchellh/go-wordwrap v1.0.1 // indirect
 	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
@@ -407,7 +408,6 @@ require (
 	github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect
 	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/temoto/robotstxt v1.1.2 // indirect
-	github.com/testcontainers/testcontainers-go v0.43.0 // indirect
 	github.com/tidwall/gjson v1.18.0 // indirect
 	github.com/tidwall/match v1.2.0 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect

pkg/jobs/canary/canary_jobs_test.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
+	"sync"
 	"time"
 
 	canaryCtx "github.com/flanksource/canary-checker/api/context"
@@ -129,3 +131,118 @@ var _ = ginkgo.Describe("Transformed Canary", ginkgo.Ordered, func() {
 		Expect(transformedCanary.DeletedAt).ToNot(BeNil())
 	})
 })
+
+var _ = ginkgo.Describe("SyncCanaryJob concurrent reschedule", ginkgo.Ordered, func() {
+	var (
+		canaryID uuid.UUID
+		specV1   types.JSON
+		specV2   types.JSON
+		dbCanary pkg.Canary
+	)
+
+	ginkgo.BeforeAll(func() {
+		canaryCtx.DefaultContext = DefaultContext
+		canaryID = uuid.New()
+
+		specV1 = types.JSON(fmt.Sprintf(`{
+			"schedule": "@every 30s",
+			"http": [{
+				"name": "concurrent-test",
+				"endpoint": "http://127.0.0.1:1/v1"
+			}]
+		}`))
+
+		specV2 = types.JSON(fmt.Sprintf(`{
+			"schedule": "@every 30s",
+			"http": [{
+				"name": "concurrent-test",
+				"endpoint": "http://127.0.0.1:1/v2"
+			}]
+		}`))
+
+		model := &models.Canary{
+			ID:        canaryID,
+			Name:      "concurrent-reschedule-test",
+			Namespace: "default",
+			AgentID:   uuid.Nil,
+			Source:    "kubernetes/" + canaryID.String(),
+			Spec:      specV1,
+		}
+		Expect(DefaultContext.DB().Create(model).Error).To(BeNil())
+
+		dbCanary = pkg.Canary{
+			ID:        canaryID,
+			Name:      model.Name,
+			Spec:      specV1,
+			Source:    model.Source,
+			Namespace: model.Namespace,
+		}
+
+		// Initial sync to populate canaryJobs map and cron.
+		Expect(SyncCanaryJob(DefaultContext, dbCanary)).To(BeNil())
+
+		// Clear any entries accumulated before this spec.
+		Unschedule(canaryID.String())
+	})
+
+	ginkgo.It("must create exactly 1 cron entry after concurrent reschedules", func() {
+		// Use spec V2 so DeepEqual detects a change.
+		dbCanaryV2 := dbCanary
+		dbCanaryV2.Spec = specV2
+
+		// Initial sync with V2 — creates a single cron entry.
+		Expect(SyncCanaryJob(DefaultContext, dbCanaryV2)).To(BeNil())
+
+		before := countCronEntriesForCanary(canaryID.String())
+		Expect(before).To(Equal(1), "expected exactly 1 cron entry after initial sync")
+
+		// Simulate concurrent reschedules.  Each goroutine changes the spec
+		// and calls SyncCanaryJob again, mimicking the race between a
+		// controller reconcile and the periodic SyncCanaryJobs job.
+		const goroutines = 10
+		var wg sync.WaitGroup
+		errs := make(chan error, goroutines)
+
+		for i := 0; i < goroutines; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				// Toggle between V1 and V2 so every call sees a
+				// "changed" spec relative to whatever is in the map.
+				c := dbCanary
+				if i%2 == 0 {
+					c.Spec = specV2
+				} else {
+					c.Spec = specV1
+				}
+				if err := SyncCanaryJob(DefaultContext, c); err != nil {
+					errs <- err
+				}
+			}(i)
+		}
+		wg.Wait()
+		close(errs)
+
+		for err := range errs {
+			ginkgo.Fail(fmt.Sprintf("unexpected error: %v", err))
+		}
+
+		// Verify only one cron entry exists for this canary.
+		after := countCronEntriesForCanary(canaryID.String())
+		Expect(after).To(Equal(1),
+			"expected exactly 1 cron entry after concurrent reschedules, got %d", after)
+	})
+})
+
+// countCronEntriesForCanary returns the number of cron entries whose
+// job carries the given canary Kubernetes UID.
+func countCronEntriesForCanary(canaryUID string) int {
+	count := 0
+	for _, entry := range CanaryScheduler.Entries() {
+		jobUID := string(entry.Job.(*job.Job).GetObjectMeta().UID)
+		if strings.EqualFold(jobUID, canaryUID) {
+			count++
+		}
+	}
+	return count
+}

pkg/jobs/canary/sync.go

@@ -24,6 +24,19 @@ import (
 
 var canaryJobs sync.Map
 
+// canarySyncLocks provides per-canary-ID mutual exclusion for SyncCanaryJob
+// to prevent a race where two concurrent callers both create cron entries for
+// the same canary, leaving an orphaned entry that fires on every tick and
+// cannot be cleaned up.
+var canarySyncLocks sync.Map
+
+func lockCanarySync(id string) func() {
+	mu := &sync.Mutex{}
+	actual, _ := canarySyncLocks.LoadOrStore(id, mu)
+	actual.(*sync.Mutex).Lock()
+	return actual.(*sync.Mutex).Unlock
+}
+
 const DefaultCanarySchedule = "@every 5m"
 
 func Unschedule(id string) {
@@ -63,6 +76,15 @@ func findJob(dbCanary pkg.Canary) *job.Job {
 
 func SyncCanaryJob(ctx context.Context, dbCanary pkg.Canary) error {
 	id := dbCanary.ID.String()
+
+	// Serialize SyncCanaryJob per canary to prevent duplicate cron entries.
+	// Without this lock, concurrent calls can race: one call Unschedule's
+	// and deletes the map entry, another sees nil and creates a new cron
+	// entry, then the first also creates a second cron entry. The orphaned
+	// entry survives all cleanup sweeps because it shares the same UID.
+	unlock := lockCanarySync(id)
+	defer unlock()
+
 	ctx.Logger.V(2).Infof("SyncCanaryJob (id=%s name=%s)", dbCanary.ID, dbCanary.Name)
 
 	if ctx.Properties().On(false, "check.*.disabled") {