CRE-6266: Fix scheduler leak (#729)

Author: cedric-cordenier

cron/trigger/trigger.go

@@ -210,7 +210,7 @@ func (s *Service) RegisterTrigger(ctx context.Context, triggerID string, metadat
 	if err != nil {
 		return nil, caperrors.NewPublicSystemError(fmt.Errorf("failed to look up fastest schedule interval: %w", err), caperrors.Internal)
 	}
-	capErr := enforceFastestSchedule(s.lggr, jobDef, limit)
+	capErr := enforceFastestSchedule(s.lggr, gocron.NewScheduler, jobDef, limit)
 	if capErr != nil {
 		return nil, capErr
 	}

cron/trigger/trigger_test.go

@@ -1229,8 +1229,79 @@ func TestEnforceFastestSchedule_NonUniformSecondsField(t *testing.T) {
 	maximumFastest := 5 * time.Second
 
 	jobDef := gocron.CronJob(schedule, true)
-	capErr := enforceFastestSchedule(logger.Nop(), jobDef, maximumFastest)
+	capErr := enforceFastestSchedule(logger.Nop(), gocron.NewScheduler, jobDef, maximumFastest)
 	require.NotNil(t, capErr, "should reject schedule with 1s gaps")
 	require.Equal(t, caperrors.LimitExceeded, capErr.Code())
 	require.Contains(t, capErr.Error(), "maximum fastest cron schedule is 5s")
 }
+
+type mockScheduler struct {
+	gocron.Scheduler
+	shutdowns int
+}
+
+func (s *mockScheduler) Shutdown() error {
+	s.shutdowns++
+	return s.Scheduler.Shutdown()
+}
+
+func mockFactory(mocks *[]*mockScheduler) schedulerFactory {
+	return func(options ...gocron.SchedulerOption) (gocron.Scheduler, error) {
+		sched, err := gocron.NewScheduler(options...)
+		if err != nil {
+			return nil, err
+		}
+		mock := &mockScheduler{Scheduler: sched}
+		*mocks = append(*mocks, mock)
+		return mock, nil
+	}
+}
+
+// TestEnforceFastestSchedule_ShutsDownTempScheduler covers a regression where the
+// temporary scheduler's Shutdown was deferred below the NewJob error check, so every
+// workflow-supplied cron string that gocron failed to parse retained a scheduler
+// goroutine for the life of the plugin. Each schedule below exercises a different
+// return path through the check; all of them must dispose of the scheduler.
+func TestEnforceFastestSchedule_ShutsDownTempScheduler(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name      string
+		schedule  string
+		shouldErr bool
+	}{
+		{name: "valid schedule", schedule: everyMinute, shouldErr: false},
+		{name: "invalid timezone", schedule: "TZ=moon * * * * *", shouldErr: true},
+		{name: "empty schedule", schedule: "", shouldErr: true},
+		{name: "not a cron schedule", schedule: "d d d d d", shouldErr: true},
+		{name: "exceeds maximum fastest", schedule: everySecond, shouldErr: true},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			var mocks []*mockScheduler
+			capErr := enforceFastestSchedule(logger.Nop(), mockFactory(&mocks), gocron.CronJob(tt.schedule, true), 5*time.Second)
+			if tt.shouldErr {
+				require.NotNil(t, capErr, "schedule %q should have been rejected", tt.schedule)
+			} else {
+				require.Nil(t, capErr)
+			}
+
+			require.Len(t, mocks, 1, "expected exactly one temporary scheduler to be created")
+			require.Equal(t, 1, mocks[0].shutdowns, "temporary scheduler was not shut down on this path")
+		})
+	}
+}
+
+func TestEnforceFastestSchedule_SchedulerConstructionFails(t *testing.T) {
+	t.Parallel()
+
+	failing := func(...gocron.SchedulerOption) (gocron.Scheduler, error) {
+		return nil, errors.New("cannot construct")
+	}
+	capErr := enforceFastestSchedule(logger.Nop(), failing, gocron.CronJob(everyMinute, true), 5*time.Second)
+	require.NotNil(t, capErr)
+	require.Equal(t, caperrors.Internal, capErr.Code())
+}

cron/trigger/utils.go

@@ -11,27 +11,38 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 )
 
-func enforceFastestSchedule(lggr logger.Logger, jobDef gocron.JobDefinition, maximumFastest time.Duration) caperrors.Error {
+// schedulerFactory constructs a gocron scheduler. Production code passes
+// gocron.NewScheduler; tests pass a double to observe the scheduler's lifecycle.
+type schedulerFactory func(options ...gocron.SchedulerOption) (gocron.Scheduler, error)
+
+// enforceFastestSchedule rejects a schedule that fires more often than maximumFastest.
+//
+// The scheduler constructor is a parameter so tests can assert the temporary scheduler
+// is always disposed of; production callers pass gocron.NewScheduler.
+func enforceFastestSchedule(lggr logger.Logger, newScheduler schedulerFactory, jobDef gocron.JobDefinition, maximumFastest time.Duration) caperrors.Error {
 	var options []gocron.SchedulerOption
 	// Use a fixed location and point in time for consistency across nodes.
 	options = append(options, gocron.WithLocation(time.UTC))
 	options = append(options, gocron.WithClock(clockwork.NewFakeClockAt(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))))
 
-	tempScheduler, err := gocron.NewScheduler(options...)
+	tempScheduler, err := newScheduler(options...)
 	if err != nil {
 		return caperrors.NewPublicSystemError(fmt.Errorf("failed to initialize temp scheduler: %w", err), caperrors.Internal)
 	}
+	// gocron starts a goroutine inside NewScheduler and Shutdown is the only thing that
+	// stops it, so this defer belongs directly under the constructor: a return between
+	// the two retains the scheduler for the life of the process.
+	defer func() {
+		if err := tempScheduler.Shutdown(); err != nil {
+			lggr.Errorw("error shutting down enforceFastestSchedule temporary scheduler", "err", err)
+		}
+	}()
+
 	tempJob, err := tempScheduler.NewJob(jobDef, gocron.NewTask(func() {}))
 	if err != nil {
 		return caperrors.NewPublicUserError(fmt.Errorf("failed to initialize job: %w", err), caperrors.InvalidArgument)
 	}
 	tempScheduler.Start()
-	defer func() {
-		err := tempScheduler.Shutdown()
-		if err != nil {
-			lggr.Errorw("error shutting down enforceFastestSchedule temporary scheduler")
-		}
-	}()
 
 	// We need to check several runs to make sure there are enough to catch any short gaps (see unit test).
 	// 12 is technically not enough in a general case but should work in practice when maximumFastest is between 5 and 60 seconds.