diff --git a/cron/cmd/cronv2/main.go b/cron/cmd/cronv2/main.go deleted file mode 100644 index 9ead895d2..000000000 --- a/cron/cmd/cronv2/main.go +++ /dev/null @@ -1,55 +0,0 @@ -package main - -import ( - "github.com/jonboulle/clockwork" - "github.com/samber/do/v2" - - "github.com/smartcontractkit/capabilities/cron/trigger" - "github.com/smartcontractkit/capabilities/libs/capability" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" -) - -func main() { - capability.Run( - "cron", - newCron, - capability.WithOtelViews(trigger.MetricViews()...), - capability.WithConfig[trigger.Config](), - ) -} - -func newCron(i do.Injector) (*server.CronServer, error) { - lggr, err := do.Invoke[logger.Logger](i) - if err != nil { - return nil, err - } - - clock, err := do.Invoke[clockwork.Clock](i) - if err != nil { - return nil, err - } - - factory, err := do.Invoke[limits.Factory](i) - if err != nil { - return nil, err - } - - cfg, err := do.Invoke[*trigger.Config](i) - if err != nil { - return nil, err - } - - triggerService, err := trigger.NewService( - lggr, - clock, - factory, - *cfg, - ) - if err != nil { - return nil, err - } - return server.NewCronServer(triggerService), nil -} diff --git a/cron/main.go b/cron/main.go deleted file mode 100644 index 42f3ea4a4..000000000 --- a/cron/main.go +++ /dev/null @@ -1,20 +0,0 @@ -package main - -import ( - "github.com/smartcontractkit/capabilities/cron/trigger" - "github.com/smartcontractkit/capabilities/libs/loopserver" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" - "github.com/smartcontractkit/chainlink-common/pkg/loop" -) - -func main() { - loopserver.ServeNew(trigger.ServiceName, func(s *loop.Server) loop.StandardCapabilities { - triggerService, err := trigger.NewLegacyService(s.Logger, nil, s.LimitsFactory) - if err != nil { - s.Logger.Fatalw("Failed to create cron trigger service", "error", err) - } - - return server.NewCronServer(triggerService) - }, loop.WithOtelViews(trigger.MetricViews())) -} diff --git a/cron/trigger/legacy.go b/cron/trigger/legacy.go deleted file mode 100644 index 1ea09773a..000000000 --- a/cron/trigger/legacy.go +++ /dev/null @@ -1,63 +0,0 @@ -package trigger - -import ( - "context" - "encoding/json" - "fmt" - - "github.com/jonboulle/clockwork" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" - "github.com/smartcontractkit/chainlink-common/pkg/types/core" -) - -// LegacyService is the Service for the LOOP plugin, where the node passes the config and the -// OrgResolver through Initialise. -type LegacyService struct { - *Service -} - -var _ server.CronCapability = &LegacyService{} - -func NewLegacyService(lggr logger.Logger, clock clockwork.Clock, limitsFactory limits.Factory) (*LegacyService, error) { - s, err := NewService(lggr, clock, limitsFactory, Config{}) - if err != nil { - return nil, err - } - return &LegacyService{Service: s}, nil -} - -func (s *LegacyService) Initialise(ctx context.Context, dependencies core.StandardCapabilitiesDependencies) error { - s.lggr.Debugw("Initialising cron trigger capability", "serviceName", ServiceName) - - if len(dependencies.Config) > 0 { - var cronConfig Config - if err := json.Unmarshal([]byte(dependencies.Config), &cronConfig); err != nil { - return fmt.Errorf("failed to unmarshal config: %s %w", dependencies.Config, err) - } - if cronConfig.FastestScheduleIntervalSeconds > 0 { - limiter, err := newFastestScheduleInterval(s.limitsFactory, cronConfig) - if err != nil { - return err - } - if err := s.fastestScheduleInterval.Close(); err != nil { - s.lggr.Warnw("Failed to close the default fastest schedule interval limiter", "err", err) - } - s.fastestScheduleInterval = limiter - } - } - - s.shardIndex = dependencies.ShardIndex - s.orgResolver = dependencies.OrgResolver - if s.orgResolver == nil { - s.lggr.Warn("OrgResolver is nil, cron capability will not be able to fetch organization ID") - } - - if err := s.Start(ctx); err != nil { - return fmt.Errorf("error when starting trigger service: %w", err) - } - - return nil -} diff --git a/cron/trigger/legacy_test.go b/cron/trigger/legacy_test.go deleted file mode 100644 index 4fda5de3b..000000000 --- a/cron/trigger/legacy_test.go +++ /dev/null @@ -1,53 +0,0 @@ -package trigger - -import ( - "testing" - - "github.com/jonboulle/clockwork" - "github.com/stretchr/testify/require" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" - "github.com/smartcontractkit/chainlink-common/pkg/types/core" -) - -func TestLegacyService_Initialise(t *testing.T) { - cases := []struct { - name string - config string - defaultClosed bool - errString string - }{ - {name: "no config keeps default limit", config: "", errString: "[101]LimitExceeded: maximum fastest cron schedule is 30s"}, - {name: "config without limit keeps default limit", config: "{}", errString: "[101]LimitExceeded: maximum fastest cron schedule is 30s"}, - {name: "config with limit replaces default limit", config: `{"fastestScheduleIntervalSeconds":1}`, defaultClosed: true}, - } - for _, tt := range cases { - t.Run(tt.name, func(t *testing.T) { - ts, err := NewLegacyService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}) - require.NoError(t, err) - defaultLimiter := &closeCountingLimiter{TimeLimiter: ts.fastestScheduleInterval} - ts.fastestScheduleInterval = defaultLimiter - - require.NoError(t, ts.Initialise(t.Context(), core.StandardCapabilitiesDependencies{Config: tt.config})) - require.Equal(t, tt.defaultClosed, defaultLimiter.closed == 1) - t.Cleanup(func() { require.NoError(t, ts.Close()) }) - - _, _, err = registerTriggerToCronTriggerService(t.Context(), t, server.NewCronServer(ts), everySecond, triggerID1) - if tt.errString != "" { - require.EqualError(t, err, tt.errString) - } else { - require.NoError(t, err) - } - }) - } -} - -func TestLegacyService_InitialiseInvalidConfig(t *testing.T) { - ts, err := NewLegacyService(logger.Nop(), clockwork.NewFakeClock(), limits.Factory{}) - require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, ts.Close()) }) - - require.ErrorContains(t, ts.Initialise(t.Context(), core.StandardCapabilitiesDependencies{Config: "{"}), "failed to unmarshal config") -} diff --git a/cron/trigger/trigger_test.go b/cron/trigger/trigger_test.go deleted file mode 100644 index 124dbd419..000000000 --- a/cron/trigger/trigger_test.go +++ /dev/null @@ -1,1296 +0,0 @@ -package trigger - -import ( - "context" - "errors" - "fmt" - "strconv" - "sync" - "testing" - "time" - - "github.com/go-co-op/gocron/v2" - "github.com/jonboulle/clockwork" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/mock" - "github.com/stretchr/testify/require" - "go.uber.org/zap" - "google.golang.org/protobuf/types/known/anypb" - - "github.com/smartcontractkit/chainlink-common/pkg/capabilities" - caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" - crontypedapi "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron" - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" - "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" - "github.com/smartcontractkit/chainlink-common/pkg/workflows" -) - -const ( - // Schedules - everyYear = "0 0 0 1 1 *" - everyMonth = "0 0 0 1 * *" - everyWeek = "0 0 0 * * 0" - everyDay = "0 0 0 * * *" - everyDayEasternTZ = "TZ=America/New_York 0 0 * * * *" - everyHour = "0 0 * * * *" - everyHourFrom9To10 = "0 9-10 * * *" - everyMinute = "0 * * * * *" - everySecond = "* * * * * *" - everyEvenSecond = "*/2 * * * * *" - - // Workflow IDs - workflowID1 = "workflow-id-1" - - // Trigger IDs - triggerID1 = "test-id-1" - triggerID2 = "test-id-2" -) - -type TriggerCapability interface { - RegisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) - UnregisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) error -} - -func registerTriggerToCronTriggerService( - ctx context.Context, - t *testing.T, - cts TriggerCapability, - schedule string, - triggerID string, -) ( - <-chan capabilities.TriggerResponse, - capabilities.TriggerRegistrationRequest, - error, -) { - requestTriggerID := workflowID1 + "|" + triggerID // TODO: remove wid once added by workflow engine - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - - payload, err := anypb.New(&crontypedapi.Config{Schedule: schedule}) - require.NoError(t, err) - - request := capabilities.TriggerRegistrationRequest{ - TriggerID: requestTriggerID, - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - triggerEventsCh, err := cts.RegisterTrigger(ctx, request) - - return triggerEventsCh, request, err -} - -// Response is the decoded form of a cron trigger event, used by these tests to -// assert on the event envelope and its payload together. -type Response struct { - capabilities.TriggerEvent - Payload Payload -} - -// Payload is a local copy of the payload type that used to live in -// chainlink-common's pkg/capabilities/triggers/cron, which has since been removed. -type Payload struct { - // Time that cron trigger's task execution had been scheduled to occur - // (RFC3339Nano formatted) - ScheduledExecutionTime string `json:"ScheduledExecutionTime" yaml:"ScheduledExecutionTime" mapstructure:"ScheduledExecutionTime"` -} - -func upwrapCronTriggerEvent(t *testing.T, event capabilities.TriggerEvent) Response { - response := Response{TriggerType: event.TriggerType} - assert.Equal(t, server.CronID, response.TriggerType) - response.ID = event.ID - - var scheduled time.Time - payload := &crontypedapi.Payload{} - require.NoError(t, event.Payload.UnmarshalTo(payload)) - scheduled = payload.ScheduledExecutionTime.AsTime() - response.Payload = Payload{ScheduledExecutionTime: scheduled.Format(time.RFC3339Nano)} - return response -} - -func makeTriggerID(number int) string { - // avoid conversion overflow for negative numbers - if number < 0 { - fmt.Printf("Trigger ID cannot be negative: %d", number) - return "" - } - return "test-id-" + strconv.FormatUint(uint64(number), 10) -} - -func requireNoChanMsg[T any](t *testing.T, ch <-chan T) { - timedOut := false - select { - case <-ch: - case <-time.After(100 * time.Millisecond): - timedOut = true - } - require.True(t, timedOut) -} - -func TestCronTrigger_SuccessWithStandardCronIntervals(t *testing.T) { - cases := []struct { - name string - schedule string - interval [5]time.Duration - }{ - { - name: "success - every second", - schedule: everySecond, - interval: [5]time.Duration{ - time.Second, - time.Second, - time.Second, - time.Second, - time.Second, - }, - }, - { - name: "success - every minute", - schedule: everyMinute, - interval: [5]time.Duration{ - time.Minute, - time.Minute, - time.Minute, - time.Minute, - time.Minute, - }, - }, - { - name: "success - every hour", - schedule: everyHour, - interval: [5]time.Duration{ - time.Hour, - time.Hour, - time.Hour, - time.Hour, - time.Hour, - }, - }, - { - name: "success - every day", - schedule: everyDay, - interval: [5]time.Duration{ - 24 * time.Hour, - 24 * time.Hour, - 24 * time.Hour, - 24 * time.Hour, - 24 * time.Hour, - }, - }, - { - name: "success - every week on Sunday", - schedule: everyWeek, - interval: [5]time.Duration{ - (time.Hour * 24) * 7, - (time.Hour * 24) * 7, - (time.Hour * 24) * 7, - (time.Hour * 24) * 7, - (time.Hour * 24) * 7, - }, - }, - { - name: "success - every month", - schedule: everyMonth, - interval: [5]time.Duration{ - (time.Hour * 24) * 31, - (time.Hour * 24) * 28, - (time.Hour * 24) * 31, - (time.Hour * 24) * 30, - (time.Hour * 24) * 31, - }, - }, - { - name: "success - every year", - schedule: everyYear, - interval: [5]time.Duration{ - (time.Hour * 24) * 365, - (time.Hour * 24) * 365, - (time.Hour * 24) * 365, - (time.Hour * 24) * 365, - (time.Hour * 24) * 365, - }, - }, - } - for _, tt := range cases { - t.Run(tt.name, func(t *testing.T) { - t.Parallel() - - startTime, _ := time.Parse("2006-01-01", "22-01-01") - fakeClock := clockwork.NewFakeClockAt(startTime) - if tt.schedule == everyWeek { - // If every week set to Saturday - for fakeClock.Now().UTC().Weekday() != time.Sunday { - fakeClock.Advance(24 * time.Hour) - } - } - - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - triggerAPI := server.NewCronServer(ts) - - // Register trigger - callback, registerUnregisterRequest, err := registerTriggerToCronTriggerService( - t.Context(), - t, - triggerAPI, - tt.schedule, - makeTriggerID(1), - ) - require.NoError(t, err) - assert.Equal(t, len(ts.scheduler.Jobs()), 1) - - // Advance to 1ms before scheduled time, there should be no channel message - fakeClock.Advance(tt.interval[0] - time.Millisecond) - requireNoChanMsg(t, callback) - // Pass scheduled time by 1ms - fakeClock.Advance(2 * time.Millisecond) - - // 1st process - msg := <-callback - response := upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime1, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - fakeClock.Advance(tt.interval[1]) - - // 2nd process - msg = <-callback - response = upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime2, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - fakeClock.Advance(tt.interval[2]) - - // 3rd process - msg = <-callback - response = upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime3, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Unregister the trigger and check that events no longer go on the callback - require.NoError(t, triggerAPI.UnregisterTrigger(t.Context(), registerUnregisterRequest)) - assert.Equal(t, len(ts.scheduler.Jobs()), 0) - assert.Equal(t, ts.scheduler.JobsWaitingInQueue(), 0) - - // Skip to when the next execution would be - fakeClock.Advance(tt.interval[3]) - - // One interval after unregistering, channel should be closed - _, open := <-callback - require.False(t, open) - - // Close the service - require.NoError(t, ts.Close()) - - // Check scheduled execution times are every interval - require.True(t, scheduledExecutionTime3.Equal(scheduledExecutionTime2.Add(tt.interval[2]))) - require.True(t, scheduledExecutionTime3.Equal(scheduledExecutionTime1.Add(tt.interval[1]+tt.interval[2]))) - require.True(t, scheduledExecutionTime2.Equal(scheduledExecutionTime1.Add(tt.interval[1]))) - }) - } -} - -func TestCronTrigger_Load(t *testing.T) { - t.Parallel() - const numTriggers = 1_000 - const numExecutions = 3 - require.Greater(t, numTriggers, 0) - require.Greater(t, numExecutions, 0) - - fakeClock := clockwork.NewRealClock() - - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - triggerAPI := server.NewCronServer(ts) - - ctx := t.Context() - - var callbacks [numTriggers]<-chan capabilities.TriggerResponse - var unregisterRequests [numTriggers]capabilities.TriggerRegistrationRequest - - // Register triggers - for triggerIdx := range numTriggers { - callback, unregisterRequest, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everySecond, - makeTriggerID(triggerIdx+1), - ) - require.NoError(t, err) - callbacks[triggerIdx] = callback - unregisterRequests[triggerIdx] = unregisterRequest - } - assert.Equal(t, len(ts.scheduler.Jobs()), numTriggers) - - // Process "numExecutions" times - var timestamps [numTriggers][numExecutions]time.Time - var scheduledExecTimes [numTriggers][numExecutions]time.Time - var mu sync.Mutex // Mutex to ensure memory is synced across threads - - wg := sync.WaitGroup{} - - for triggerIdx := range numTriggers { - wg.Add(1) - go func(tIdx int) { - for execIdx := range numExecutions { - msg := <-callbacks[tIdx] - response := upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime, _ := time.Parse(time.RFC3339Nano, response.Payload.ScheduledExecutionTime) - - actualExecutionTimeUTC := fakeClock.Now().UTC() - - mu.Lock() - scheduledExecTimes[tIdx][execIdx] = scheduledExecutionTime - timestamps[tIdx][execIdx] = actualExecutionTimeUTC - mu.Unlock() - } - wg.Done() - }(triggerIdx) - } - - wg.Wait() - - mu.Lock() - defer mu.Unlock() - - // Unregister the trigger and check that events no longer go on the callback - for i := range numTriggers { - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, unregisterRequests[i])) - } - - assert.Equal(t, len(ts.scheduler.Jobs()), 0) - assert.Equal(t, ts.scheduler.JobsWaitingInQueue(), 0) - -triggers: - for i := range numTriggers { - closeTimer := time.NewTimer(10 * time.Second) - for { - select { - case _, ok := <-callbacks[i]: - if !ok { - closeTimer.Stop() - continue triggers - } - case <-closeTimer.C: - t.Fatalf("timed out waiting for channel %d to close", i) - } - } - } - - // Close the service - require.NoError(t, ts.Close()) - - var scheduledActualDelta [numTriggers * numExecutions]int64 - - for execIdx := range numExecutions { - var earliestScheduled time.Time - var latestScheduled time.Time - - for triggerIdx := range numTriggers { - // Triggers with the same schedule should remain aligned, but real-clock - // registration can span a second boundary under load. - if triggerIdx == 0 { - earliestScheduled = scheduledExecTimes[triggerIdx][execIdx] - latestScheduled = scheduledExecTimes[triggerIdx][execIdx] - } else { - if scheduledExecTimes[triggerIdx][execIdx].Before(earliestScheduled) { - earliestScheduled = scheduledExecTimes[triggerIdx][execIdx] - } - if scheduledExecTimes[triggerIdx][execIdx].After(latestScheduled) { - latestScheduled = scheduledExecTimes[triggerIdx][execIdx] - } - } - // Check that executions happened every second - if execIdx > 0 { - require.True(t, scheduledExecTimes[triggerIdx][execIdx].Equal(scheduledExecTimes[triggerIdx][execIdx-1].Add(time.Second))) - } - // Check that actual execution time is after scheduled time - - after := timestamps[triggerIdx][execIdx].After(scheduledExecTimes[triggerIdx][execIdx]) || timestamps[triggerIdx][execIdx].Equal(scheduledExecTimes[triggerIdx][execIdx]) - - require.True(t, after) - - // Check that scheduled time and actual time did not differ more than 1 second - require.False(t, timestamps[triggerIdx][execIdx].After(scheduledExecTimes[triggerIdx][execIdx].Add(time.Second))) - // Store time difference between scheduled and actual - scheduledActualDelta[triggerIdx*numExecutions+execIdx] = timestamps[triggerIdx][execIdx].Sub(scheduledExecTimes[triggerIdx][execIdx]).Milliseconds() - } - - require.False(t, latestScheduled.After(earliestScheduled.Add(time.Second))) - } - - var averageDelta int64 - for _, num := range scheduledActualDelta { - averageDelta += num - } - averageDelta = averageDelta / int64(len(scheduledActualDelta)) - t.Logf("Average Delta: %dms\n", averageDelta) -} - -func TestCronTrigger_RegisterTriggerBeforeStart(t *testing.T) { - fakeClock := clockwork.NewRealClock() - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - triggerAPI := server.NewCronServer(ts) - - ctx := t.Context() - - // Register trigger - callback, registerUnregisterRequest, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everySecond, - makeTriggerID(1), - ) - require.NoError(t, err) - assert.Equal(t, len(ts.scheduler.Jobs()), 1) - - // 1st process - msg := <-callback - actualExecutionTime1 := fakeClock.Now().UTC() - response := upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime1, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // 2nd process - msg = <-callback - actualExecutionTime2 := fakeClock.Now().UTC() - response = upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime2, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Unregister the trigger and check that events no longer go on the callback - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, registerUnregisterRequest)) - assert.Equal(t, len(ts.scheduler.Jobs()), 0) - assert.Equal(t, ts.scheduler.JobsWaitingInQueue(), 0) - - // Close the service - require.NoError(t, ts.Close()) - - // Check that executions happened every second - require.True(t, scheduledExecutionTime2.Equal(scheduledExecutionTime1.Add(time.Second))) - // Check that actual execution time is after scheduled time - require.True(t, actualExecutionTime1.After(scheduledExecutionTime1)) - require.True(t, actualExecutionTime2.After(scheduledExecutionTime2)) - // Check that scheduled time and actual time did not differ more than 1 second - require.False(t, actualExecutionTime1.After(scheduledExecutionTime1.Add(time.Second))) - require.False(t, actualExecutionTime2.After(scheduledExecutionTime2.Add(time.Second))) -} - -func TestCronTriggerTimeWindows(t *testing.T) { - fakeClock := clockwork.NewFakeClock() - // Set time to have 0ms by advancing to next truncated second - fakeClock.Advance(fakeClock.Now().Truncate(time.Second).Add(time.Second).Sub(fakeClock.Now())) - // Set time to 8:50am UTC - hour, minimum, sec := fakeClock.Now().UTC().Clock() - fakeClock.Advance(time.Duration(60-sec) * time.Second) - fakeClock.Advance(time.Duration(49-minimum) * time.Minute) - fakeClock.Advance(time.Duration(8-hour) * time.Hour) - - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - - ctx := t.Context() - - // Register trigger - callback, registerUnregisterRequest, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everyHourFrom9To10, - makeTriggerID(1), - ) - require.NoError(t, err) - assert.Equal(t, len(ts.scheduler.Jobs()), 1) - - // Advance to 1ms past 9am - fakeClock.Advance(10*time.Minute + time.Millisecond) - - // 1st process @ 9am UTC - msg := <-callback - response := upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime1, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Advance to 10am - fakeClock.Advance(time.Hour) - - // 2nd process @ 10am UTC - msg = <-callback - response = upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime2, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Advance to 9am UTC next day - fakeClock.Advance(time.Hour * 23) - - // should not process again until next day - msg = <-callback - response = upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime3, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Unregister the trigger and check that events no longer go on the callback - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, registerUnregisterRequest)) - assert.Equal(t, len(ts.scheduler.Jobs()), 0) - assert.Equal(t, ts.scheduler.JobsWaitingInQueue(), 0) - - // Close the service - require.NoError(t, ts.Close()) - - // Check scheduled execution times 9am, 10am, then 9am the next day - require.True(t, scheduledExecutionTime3.Equal(scheduledExecutionTime2.Add(23*time.Hour))) - require.True(t, scheduledExecutionTime3.Equal(scheduledExecutionTime1.Add(24*time.Hour))) - require.True(t, scheduledExecutionTime2.Equal(scheduledExecutionTime1.Add(time.Hour))) -} - -func TestCronTrigger_MultipleDifferentSchedules(t *testing.T) { - fakeClock := clockwork.NewFakeClock() - // Start on an odd numbered second - if fakeClock.Now().Second()%2 == 1 { - fakeClock.Advance(time.Second) - } - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - ctx := t.Context() - - callback1, registerUnregisterRequest1, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everySecond, - triggerID1, - ) - require.NoError(t, err) - - callback2, registerUnregisterRequest2, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everyEvenSecond, - triggerID2, - ) - require.NoError(t, err) - - assert.Equal(t, len(ts.scheduler.Jobs()), 2) - - fakeClock.Advance(time.Second) - - // 1st second - msg1 := <-callback1 - response1 := upwrapCronTriggerEvent(t, msg1.Event) - scheduledExecutionTime1_1, _ := time.Parse(time.RFC3339, response1.Payload.ScheduledExecutionTime) - - fakeClock.Advance(time.Second) - - // 2nd second - msg1 = <-callback1 - response1 = upwrapCronTriggerEvent(t, msg1.Event) - scheduledExecutionTime1_2, _ := time.Parse(time.RFC3339, response1.Payload.ScheduledExecutionTime) - eventID1Run2 := response1.ID - - msg2 := <-callback2 - response2 := upwrapCronTriggerEvent(t, msg2.Event) - scheduledExecutionTime2_1, _ := time.Parse(time.RFC3339, response2.Payload.ScheduledExecutionTime) - eventID2Run2 := response2.ID - - fakeClock.Advance(time.Second) - - // 3rd second - msg1 = <-callback1 - response1 = upwrapCronTriggerEvent(t, msg1.Event) - scheduledExecutionTime1_3, _ := time.Parse(time.RFC3339, response1.Payload.ScheduledExecutionTime) - - fakeClock.Advance(time.Second) - - // 4th second - msg1 = <-callback1 - response1 = upwrapCronTriggerEvent(t, msg1.Event) - scheduledExecutionTime1_4, _ := time.Parse(time.RFC3339, response1.Payload.ScheduledExecutionTime) - eventID1Run4 := response1.ID - - msg2 = <-callback2 - response2 = upwrapCronTriggerEvent(t, msg2.Event) - scheduledExecutionTime2_2, _ := time.Parse(time.RFC3339, response2.Payload.ScheduledExecutionTime) - eventID2Run4 := response2.ID - - // Unregister the trigger and check that events no longer go on the callback - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, registerUnregisterRequest1)) - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, registerUnregisterRequest2)) - - fakeClock.Advance(time.Second) - - _, open := <-callback1 - require.False(t, open) - - _, open = <-callback2 - require.False(t, open) - - // Close the service - require.NoError(t, ts.Close()) - - // Check scheduled execution times - // Trigger 1 happened every second - require.True(t, scheduledExecutionTime1_4.Equal(scheduledExecutionTime1_3.Add(time.Second))) - require.True(t, scheduledExecutionTime1_4.Equal(scheduledExecutionTime1_2.Add(time.Second*2))) - require.True(t, scheduledExecutionTime1_4.Equal(scheduledExecutionTime1_1.Add(time.Second*3))) - require.True(t, scheduledExecutionTime1_3.Equal(scheduledExecutionTime1_2.Add(time.Second))) - require.True(t, scheduledExecutionTime1_3.Equal(scheduledExecutionTime1_1.Add(time.Second*2))) - require.True(t, scheduledExecutionTime1_2.Equal(scheduledExecutionTime1_1.Add(time.Second))) - // Trigger 2 happened every second second - require.True(t, scheduledExecutionTime2_2.Equal(scheduledExecutionTime2_1.Add(time.Second*2))) - // The 2nd and 4th second have the same event ID - require.Equal(t, eventID1Run2, eventID2Run2) - require.Equal(t, eventID1Run4, eventID2Run4) -} - -func TestCronTriggerTimeZone(t *testing.T) { - fakeClock := clockwork.NewFakeClock() - location, _ := time.LoadLocation("America/New_York") - // Set time to have 0ms by advancing to next truncated second - fakeClock.Advance(time.Duration(1000000000 - fakeClock.Now().Nanosecond())) - // Set time to 23:50pm Eastern - now := fakeClock.Now().In(location) - hour, minimum, sec := now.Clock() - fakeClock.Advance(time.Duration(60-sec) * time.Second) - fakeClock.Advance(time.Duration(49-minimum) * time.Minute) - fakeClock.Advance(time.Duration(23-hour) * time.Hour) - - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - ctx := t.Context() - - // Register trigger - callback, registerUnregisterRequest, err := registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - everyDayEasternTZ, - makeTriggerID(1), - ) - require.NoError(t, err) - assert.Equal(t, len(ts.scheduler.Jobs()), 1) - - // Advance to 1ms before trigger - fakeClock.Advance(9*time.Minute + 59*time.Second + 999*time.Millisecond) - - // There should be no channel message - requireNoChanMsg(t, callback) - - // Advance to next 12am Eastern - fakeClock.Advance(time.Millisecond) - - // 1st process @ 12am Eastern - msg := <-callback - response := upwrapCronTriggerEvent(t, msg.Event) - scheduledExecutionTime, _ := time.Parse(time.RFC3339, response.Payload.ScheduledExecutionTime) - - // Unregister the trigger and check that events no longer go on the callback - require.NoError(t, triggerAPI.UnregisterTrigger(ctx, registerUnregisterRequest)) - assert.Equal(t, len(ts.scheduler.Jobs()), 0) - assert.Equal(t, ts.scheduler.JobsWaitingInQueue(), 0) - - // Close the service - require.NoError(t, ts.Close()) - - // Check scheduled execution is at 12am Eastern - timezone, err := time.LoadLocation("America/New_York") - require.NoError(t, err) - expectedEasternExecution := time.Date(now.Year(), now.Month(), now.Day()+1, 0, 0, 0, 0, timezone) - require.True(t, scheduledExecutionTime.Equal(expectedEasternExecution)) -} - -func TestCronTrigger_RegisterTrigger(t *testing.T) { - cases := []struct { - name string - schedule string - shouldErr bool - expectedErrString string - errorOrigin caperrors.Origin - }{ - // No Error - { - name: "valid cron schedule - 6 entries", - schedule: everyMinute, - shouldErr: false, - expectedErrString: "", - }, - { - name: "valid cron schedule - 5 entries", - schedule: "* * * * *", - shouldErr: false, - expectedErrString: "", - }, - - // Error - { - name: "invalid cron schedule - empty", - schedule: "", - shouldErr: true, - expectedErrString: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nexpected 5 to 6 fields, found 0: []", - errorOrigin: caperrors.OriginUser, - }, - { - name: "invalid cron schedule - not a cron schedule", - schedule: "d d d d d", - shouldErr: true, - expectedErrString: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nfailed to parse int from d: strconv.Atoi: parsing \"d\": invalid syntax", - errorOrigin: caperrors.OriginUser, - }, - { - name: "invalid cron schedule - invalid timezone", - schedule: "TZ=moon * * * * *", - shouldErr: true, - expectedErrString: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nprovided bad location moon: unknown time zone moon", - errorOrigin: caperrors.OriginUser, - }, - { - name: "invalid cron schedule - exceeds maximum fastest", - schedule: everySecond, - shouldErr: true, - expectedErrString: "[101]LimitExceeded: maximum fastest cron schedule is 30s", - errorOrigin: caperrors.OriginUser, - }, - } - for _, tt := range cases { - t.Run(tt.name, func(t *testing.T) { - fakeClock := clockwork.NewRealClock() - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - ctx := t.Context() - _, _, err = registerTriggerToCronTriggerService( - ctx, - t, - triggerAPI, - tt.schedule, - triggerID1, - ) - - if tt.shouldErr { - require.Error(t, err) - if tt.expectedErrString != "" { - require.Equal(t, tt.expectedErrString, err.Error()) - } - - var capError caperrors.Error - require.True(t, errors.As(err, &capError)) - - require.Equal(t, tt.errorOrigin, capError.Origin()) - require.Equal(t, caperrors.VisibilityPublic, capError.Visibility()) - } else { - require.NoError(t, err) - } - - require.NoError(t, ts.Close()) - }) - } -} - -func TestCronTrigger_RegisterTriggerDuplicateError(t *testing.T) { - fakeClock := clockwork.NewRealClock() - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - - ctx := t.Context() - - payload, err := anypb.New(&crontypedapi.Config{Schedule: everySecond}) - require.NoError(t, err) - - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - request := capabilities.TriggerRegistrationRequest{ - TriggerID: triggerID1, - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - - _, err = triggerAPI.RegisterTrigger(ctx, request) - require.NoError(t, err) - _, err = triggerAPI.RegisterTrigger(ctx, request) - require.Error(t, err) - require.Equal(t, "[13]Internal: triggerId test-id-1 already registered", err.Error()) -} - -func TestCronTrigger_UnregisterTriggerError(t *testing.T) { - fakeClock := clockwork.NewRealClock() - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - triggerAPI := server.NewCronServer(ts) - - t.Run("OK if trigger not found", func(t *testing.T) { - ctx := t.Context() - - payload, err := anypb.New(&crontypedapi.Config{Schedule: everySecond}) - require.NoError(t, err) - - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - request := capabilities.TriggerRegistrationRequest{ - TriggerID: "missing", - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - - err = triggerAPI.UnregisterTrigger(ctx, request) - require.NoError(t, err) - }) - - t.Run("OK register then unregister", func(t *testing.T) { - ctx := t.Context() - - payload, err := anypb.New(&crontypedapi.Config{Schedule: everySecond}) - require.NoError(t, err) - - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - request := capabilities.TriggerRegistrationRequest{ - TriggerID: "trigger-id", - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - - _, err = triggerAPI.RegisterTrigger(ctx, request) - require.NoError(t, err) - err = triggerAPI.UnregisterTrigger(ctx, request) - require.NoError(t, err) - }) - - t.Run("OK register then unregister multiple times", func(t *testing.T) { - ctx := t.Context() - - payload, err := anypb.New(&crontypedapi.Config{Schedule: everySecond}) - require.NoError(t, err) - - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - request := capabilities.TriggerRegistrationRequest{ - TriggerID: "trigger-id", - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - - _, err = triggerAPI.RegisterTrigger(ctx, request) - require.NoError(t, err) - - for range 3 { - err = triggerAPI.UnregisterTrigger(ctx, request) - require.NoError(t, err) - } - }) - - t.Run("NOK fails to unregister if closed", func(t *testing.T) { - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - triggerAPI := server.NewCronServer(ts) - ctx := t.Context() - - payload, err := anypb.New(&crontypedapi.Config{Schedule: everySecond}) - require.NoError(t, err) - - requestMetadata := capabilities.RequestMetadata{ - WorkflowID: workflowID1, - } - request := capabilities.TriggerRegistrationRequest{ - TriggerID: "trigger-id", - Metadata: requestMetadata, - Payload: payload, - Method: "Trigger", - } - - _, err = triggerAPI.RegisterTrigger(ctx, request) - require.NoError(t, err) - - err = triggerAPI.Close() - require.NoError(t, err) - - err = triggerAPI.UnregisterTrigger(ctx, request) - require.Error(t, err) - require.ErrorContains(t, err, "cannot unregister a new trigger, service has been closed") - }) -} - -func TestCronTrigger_ReadyAndHealthFollowLifecycle(t *testing.T) { - ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, Config{}) - require.NoError(t, err) - - require.Error(t, ts.Ready()) - require.Error(t, ts.HealthReport()[ts.Name()]) - - require.NoError(t, ts.Start(t.Context())) - require.NoError(t, ts.Ready()) - require.NoError(t, ts.HealthReport()[ts.Name()]) - - require.NoError(t, ts.Close()) - require.Error(t, ts.Ready()) - require.Error(t, ts.HealthReport()[ts.Name()]) -} - -func TestCronTrigger_CloseBeforeStart(t *testing.T) { - ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, Config{}) - require.NoError(t, err) - - require.NoError(t, ts.Close()) -} - -func TestCronTrigger_CloseStartErrors(t *testing.T) { - fakeClock := clockwork.NewRealClock() - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{}) - require.NoError(t, err) - ctx := t.Context() - - err = ts.Start(ctx) - require.NoError(t, err) - err = ts.Close() - require.NoError(t, err) - err = ts.Start(ctx) - require.Error(t, err) -} - -type panicOnNowClock struct { - clockwork.Clock -} - -func (c *panicOnNowClock) Now() time.Time { - panic("clock panic in Now()") -} - -func TestGocronNewTaskPanic(t *testing.T) { - fakeClock := clockwork.NewFakeClock() - - logger, observedLogs := logger.TestObserved(t, zap.ErrorLevel) - ts, err := NewService(logger, fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - triggerAPI := server.NewCronServer(ts) - - _, _, err = registerTriggerToCronTriggerService( - t.Context(), - t, - triggerAPI, - everySecond, - makeTriggerID(1), - ) - require.NoError(t, err) - assert.Equal(t, len(ts.scheduler.Jobs()), 1) - - // Swap ts.clock after setup so gocron's scheduler/executor keep using the - // original fakeClock, while the task callback (trigger.go s.clock.Now()) - // hits the panicking clock. - ts.clock = &panicOnNowClock{Clock: fakeClock} - - fakeClock.Advance(time.Second * 1) - - ticker := time.NewTicker(time.Second * 1) - timeout := time.NewTimer(time.Second * 10) - for { - select { - case <-ticker.C: - logs := observedLogs.All() - for _, log := range logs { - if log.Message == "panic in gocron.NewTask function" && len(log.Context) > 0 && log.Context[0].Key == "err" && log.Context[0].String == "clock panic in Now()" { - return - } - } - case <-timeout.C: - t.Fatal("timeout, no panic in gocron.NewTask function") - } - } -} - -// MockOrgResolver is a mock implementation of the OrgResolver interface -type MockOrgResolver struct { - mock.Mock -} - -func (m *MockOrgResolver) Get(ctx context.Context, owner string) (string, error) { - args := m.Called(ctx, owner) - return args.String(0), args.Error(1) -} - -func (m *MockOrgResolver) Start(ctx context.Context) error { - return nil -} - -func (m *MockOrgResolver) Close() error { - return nil -} - -func (m *MockOrgResolver) HealthReport() map[string]error { - return map[string]error{} -} - -func (m *MockOrgResolver) Ready() error { - return nil -} - -func (m *MockOrgResolver) Name() string { - return "MockOrgResolver" -} - -// Ensure MockOrgResolver implements the interface -var _ orgresolver.OrgResolver = (*MockOrgResolver)(nil) - -// TestCronTrigger_ExecutionIDWithTriggerIndex verifies execution IDs include the -// trigger index parsed from ReferenceID. -func TestCronTrigger_ExecutionIDWithTriggerIndex(t *testing.T) { - t.Parallel() - - const testWorkflowID = "workflow-id-multi-trigger-test" - const testReferenceID = "trigger_2" // parsed to triggerIndex = 2 - const testTriggerIndex = 2 - const testTriggerID = testWorkflowID + "|test-trigger" - - startTime := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) - fakeClock := clockwork.NewFakeClockAt(startTime) - lggr, observedLogs := logger.TestObserved(t, zap.DebugLevel) - - ts, err := NewService(lggr, fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - metadata := capabilities.RequestMetadata{ - WorkflowID: testWorkflowID, - ReferenceID: testReferenceID, - } - ch, capErr := ts.RegisterTrigger(t.Context(), testTriggerID, metadata, &crontypedapi.Config{Schedule: everySecond}) - require.Nil(t, capErr) - - fakeClock.Advance(time.Second + time.Millisecond) - msg := <-ch - - expectedExecID, err := workflows.GenerateExecutionIDWithTriggerIndex(testWorkflowID, msg.Id, testTriggerIndex) - require.NoError(t, err) - - // The debug log at "sending trigger event" is written before the channel - // send, so it is already present once we receive msg. - var execIDFromLog string - var isLegacyFromLog bool - var found bool - for _, entry := range observedLogs.All() { - if entry.Message == "sending trigger event" { - for _, field := range entry.Context { - if field.Key == "executionID" { - execIDFromLog = field.String - } - } - found = true - break - } - } - require.True(t, found, "expected log entry 'sending trigger event'") - require.Equal(t, expectedExecID, execIDFromLog, "execution ID should match expected hash function") - require.False(t, isLegacyFromLog) - - require.Nil(t, ts.UnregisterTrigger(t.Context(), testTriggerID, metadata, &crontypedapi.Config{Schedule: everySecond})) - require.NoError(t, ts.Close()) -} - -// TestCronTrigger_Deterministic covers every event carrying its own run's scheduled time, tick after -// tick, which is what lets every node in a DON agree on each event's ID. -func TestCronTrigger_Deterministic(t *testing.T) { - const ticks = 100 - - start := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) - fakeClock := clockwork.NewFakeClockAt(start) - - ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{FastestScheduleIntervalSeconds: 1}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - metadata := capabilities.RequestMetadata{WorkflowID: workflowID1} - input := &crontypedapi.Config{Schedule: everySecond} - ch, capErr := ts.RegisterTrigger(t.Context(), triggerID1, metadata, input) - require.Nil(t, capErr) - - for i := 1; i <= ticks; i++ { - fakeClock.Advance(time.Second) - - var e capabilities.TriggerAndId[*crontypedapi.Payload] - select { - case e = <-ch: - case <-time.After(10 * time.Second): - t.Fatalf("tick %d: timed out waiting for an event", i) - } - - want := start.Add(time.Duration(i) * time.Second) - require.Equal(t, want, e.Trigger.ScheduledExecutionTime.AsTime(), "tick %d", i) - require.Equal(t, want.Format(time.RFC3339), e.Id, "tick %d", i) - } - - require.Nil(t, ts.UnregisterTrigger(t.Context(), triggerID1, metadata, input)) - require.NoError(t, ts.Close()) -} - -// TestCronTrigger_NoLogsAfterClose covers gocron, which keeps logging for a moment after Shutdown returns: -// none of it may reach the service's logger once Close has returned. -func TestCronTrigger_NoLogsAfterClose(t *testing.T) { - for range 20 { - lggr, observedLogs := logger.TestObserved(t, zap.DebugLevel) - ts, err := NewService(lggr, clockwork.NewFakeClock(), limits.Factory{}, Config{}) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - - require.NoError(t, ts.Close()) - closed := observedLogs.Len() - - time.Sleep(10 * time.Millisecond) - require.Equal(t, closed, observedLogs.Len(), "logged after Close returned: %v", observedLogs.All()[closed:]) - } -} - -func TestEnforceFastestSchedule_NonUniformSecondsField(t *testing.T) { - t.Parallel() - // a schedule that has a bunch of 5s gaps followed by a bunch of 1s gaps - // requires checking multiple runs to catch the short gaps - schedule := "0,5,10,15,20,25,30,35,40,45,50,55-59 * * * * *" - maximumFastest := 5 * time.Second - - jobDef := gocron.CronJob(schedule, true) - 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()) -} - -func TestCronTrigger_FastestScheduleFromConstructor(t *testing.T) { - cases := []struct { - name string - cfg Config - errString string - }{ - {name: "default limit", cfg: Config{}, errString: "[101]LimitExceeded: maximum fastest cron schedule is 30s"}, - {name: "configured limit", cfg: Config{FastestScheduleIntervalSeconds: 1}}, - } - for _, tt := range cases { - t.Run(tt.name, func(t *testing.T) { - ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, tt.cfg) - require.NoError(t, err) - require.NoError(t, ts.Start(t.Context())) - t.Cleanup(func() { require.NoError(t, ts.Close()) }) - - _, _, err = registerTriggerToCronTriggerService(t.Context(), t, server.NewCronServer(ts), everySecond, triggerID1) - if tt.errString != "" { - require.EqualError(t, err, tt.errString) - } else { - require.NoError(t, err) - } - }) - } -} - -type closeCountingLimiter struct { - limits.TimeLimiter - closed int -} - -func (l *closeCountingLimiter) Close() error { - l.closed++ - return l.TimeLimiter.Close() -} - -func TestService_CloseClosesLimiter(t *testing.T) { - ts, err := NewService(logger.Nop(), clockwork.NewFakeClock(), limits.Factory{}, Config{}) - require.NoError(t, err) - limiter := &closeCountingLimiter{TimeLimiter: ts.fastestScheduleInterval} - ts.fastestScheduleInterval = limiter - require.NoError(t, ts.Start(t.Context())) - - require.NoError(t, ts.Close()) - require.Equal(t, 1, limiter.closed) -} diff --git a/go.md b/go.md index 7d8737141..b0355a71a 100644 --- a/go.md +++ b/go.md @@ -19,8 +19,6 @@ flowchart LR capabilities/consensus --> capabilities/libs capabilities/consensus --> cre-sdk-go click capabilities/consensus href "https://github.com/smartcontractkit/capabilities" - capabilities/cron --> capabilities/libs - click capabilities/cron href "https://github.com/smartcontractkit/capabilities" capabilities/http_action --> capabilities/libs click capabilities/http_action href "https://github.com/smartcontractkit/capabilities" capabilities/http_trigger --> capabilities/libs @@ -33,6 +31,8 @@ flowchart LR click capabilities/integration_tests href "https://github.com/smartcontractkit/capabilities" capabilities/libs --> chainlink-common/x/config click capabilities/libs href "https://github.com/smartcontractkit/capabilities" + capabilities/scheduler/cron --> capabilities/libs + click capabilities/scheduler/cron href "https://github.com/smartcontractkit/capabilities" ccip-contract-examples/chains/evm click ccip-contract-examples/chains/evm href "https://github.com/smartcontractkit/ccip-contract-examples" ccip-owner-contracts @@ -240,11 +240,11 @@ flowchart LR capabilities/chain_capabilities/solana capabilities/chain_capabilities/stellar capabilities/consensus - capabilities/cron capabilities/http_action capabilities/http_trigger capabilities/integration_tests capabilities/libs + capabilities/scheduler/cron end click capabilities-repo href "https://github.com/smartcontractkit/capabilities" diff --git a/libs/capability/capability.go b/libs/capability/capability.go index c2c175473..efabc996d 100644 --- a/libs/capability/capability.go +++ b/libs/capability/capability.go @@ -30,6 +30,9 @@ type CapabilitiesConfig struct { ProxyURL string // CapabilityDonID is the on-chain DON ID of the capability DON this process was spawned for. CapabilityDonID uint32 + // ShardIndex is the shard of the node this process was spawned for, from the node's [Sharding] config. Zero for + // unsharded nodes. + ShardIndex uint32 // HTTPDebug serves the capability debug UI on the shared HTTP server, under /debug/capabilities. HTTPDebug bool } diff --git a/libs/capability/doccomments_gen.go b/libs/capability/doccomments_gen.go index b8b754fc6..98f41735f 100644 --- a/libs/capability/doccomments_gen.go +++ b/libs/capability/doccomments_gen.go @@ -15,6 +15,7 @@ func (CapabilitiesConfig) DocComments() map[string]commentparsing.FieldDoc { "CapabilityDonID": {Comment: "CapabilityDonID is the on-chain DON ID of the capability DON this process was spawned for."}, "HTTPDebug": {Comment: "HTTPDebug serves the capability debug UI on the shared HTTP server, under /debug/capabilities."}, "ProxyURL": {Comment: "ProxyURL is the gRPC target of the node's capability registry proxy (e.g. localhost:9000), used\nto resolve capabilities this binary does not host. Unset resolves only the capabilities this\nbinary hosts."}, + "ShardIndex": {Comment: "ShardIndex is the shard of the node this process was spawned for, from the node's [Sharding] config. Zero for\nunsharded nodes."}, } } diff --git a/libs/capability/embed.go b/libs/capability/embed.go index 3f578ee75..e180920c1 100644 --- a/libs/capability/embed.go +++ b/libs/capability/embed.go @@ -115,6 +115,7 @@ func startCapability[T Capability](ctx context.Context, lggr logger.Logger, i *I if err := errors.Join( i.ProvideValue(info), + i.ProvideValue(cfg.capabilities), i.ProvideValue(lggr), i.ProvideValue[registry.CapabilitiesRegistry](reg), i.ProvideValue(newLimitsFactory(lggr, cfg.settings)), diff --git a/libs/capability/run_test.go b/libs/capability/run_test.go index 26d5f34da..b53fc0ecf 100644 --- a/libs/capability/run_test.go +++ b/libs/capability/run_test.go @@ -232,6 +232,18 @@ func TestRunProvidesItsDependencies(t *testing.T) { assert.NotNil(t, gotClock) } +// TestRunProvidesTheCapabilitiesConfig covers what the node spawned this process with, such as its shard. +func TestRunProvidesTheCapabilitiesConfig(t *testing.T) { + var got CapabilitiesConfig + + require.NoError(t, execute(t, logger.Test(t), func(i do.Injector) (Capability, error) { + got = do.MustInvoke[CapabilitiesConfig](i) + return newFake(), nil + }, "run", "--capabilities.shard-index", "3")) + + assert.Equal(t, uint32(3), got.ShardIndex) +} + // TestBareCommandRunsInCompatibilityMode covers a bare , which is how the core node starts // a LOOP. The test can't reach plugin.Serve, because it takes over the stdio of the process. So // each case must fail before the capability is built. diff --git a/libs/migration/migration.go b/libs/migration/migration.go new file mode 100644 index 000000000..8ea4ee8ec --- /dev/null +++ b/libs/migration/migration.go @@ -0,0 +1,144 @@ +// Package migration serves a capability built for capability.Run as a StandardCapabilities LOOP, so nodes that +// still run capabilities that way can run it while it moves to capability.Run. +package migration + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "go.opentelemetry.io/otel/attribute" + + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/libs/loopserver" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry" + capmon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/monitoring" + "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink-common/pkg/types/core" +) + +// Serve serves the capability newCapability builds as a StandardCapabilities LOOP named name. newCapability is +// called when the node calls Initialise, with what the node gives. +func Serve(name string, newCapability func(*loop.Server, core.StandardCapabilitiesDependencies) (capability.Capability, error), + opts ...loop.ServerOpt) { + loopserver.ServeNew(name, func(s *loop.Server) *standardCapabilities { + return newStandardCapabilities(name, func(deps core.StandardCapabilitiesDependencies) (capability.Capability, error) { + return newCapability(s, deps) + }) + }, opts...) +} + +var _ loop.StandardCapabilities = (*standardCapabilities)(nil) + +// standardCapabilities is a capability as a StandardCapabilities LOOP. Until Initialise, there is no capability, +// because what it is built from comes with Initialise. +type standardCapabilities struct { + name string + newCapability func(core.StandardCapabilitiesDependencies) (capability.Capability, error) + initMetrics capmon.InitMetrics + + mu sync.Mutex + capability capability.Capability + registry registry.CapabilitiesRegistry +} + +// errNotInitialised is what is answered for the capability before the node calls Initialise. +var errNotInitialised = errors.New("the capability is built when the node calls Initialise, which it has not") + +func newStandardCapabilities(name string, + newCapability func(core.StandardCapabilitiesDependencies) (capability.Capability, error)) *standardCapabilities { + return &standardCapabilities{name: name, newCapability: newCapability, initMetrics: capmon.NewInitMetrics()} +} + +// Initialise builds the capability, starts it and adds it to the node's registry, which is what the node runs it +// through. +func (s *standardCapabilities) Initialise(ctx context.Context, deps core.StandardCapabilitiesDependencies) (err error) { + defer func() { s.initMetrics.RecordInit(ctx, err, attribute.String("capability", s.name)) }() + + c, err := s.newCapability(deps) + if err != nil { + return fmt.Errorf("failed to instantiate capability: %w", err) + } + + if err = c.Start(ctx); err != nil { + return fmt.Errorf("failed to start capability: %w", err) + } + + if err = deps.CapabilityRegistry.Add(ctx, c); err != nil { + return errors.Join(fmt.Errorf("failed to add the capability to the registry: %w", err), c.Close()) + } + + s.mu.Lock() + s.capability = c + s.registry = deps.CapabilityRegistry + s.mu.Unlock() + return nil +} + +func (s *standardCapabilities) Infos(ctx context.Context) ([]capabilities.CapabilityInfo, error) { + c, _ := s.current() + if c == nil { + return nil, errNotInitialised + } + + info, err := c.Info(ctx) + if err != nil { + return nil, err + } + + return []capabilities.CapabilityInfo{info}, nil +} + +func (s *standardCapabilities) Name() string { return s.name } + +// Start does nothing: the node starts the capability with Initialise. +func (s *standardCapabilities) Start(context.Context) error { return nil } + +func (s *standardCapabilities) HealthReport() map[string]error { + c, _ := s.current() + if c == nil { + return map[string]error{s.name: errNotInitialised} + } + + return c.HealthReport() +} + +func (s *standardCapabilities) Ready() error { + c, _ := s.current() + if c == nil { + return errNotInitialised + } + + return c.Ready() +} + +func (s *standardCapabilities) current() (capability.Capability, registry.CapabilitiesRegistry) { + s.mu.Lock() + defer s.mu.Unlock() + return s.capability, s.registry +} + +// Close removes the capability from the node's registry, then closes it. +func (s *standardCapabilities) Close() error { + c, reg := s.current() + if c == nil { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + var errs []error + info, err := c.Info(ctx) + if err != nil { + errs = append(errs, fmt.Errorf("failed to read the capability's info: %w", err)) + } else if err = reg.Remove(ctx, info.ID); err != nil { + errs = append(errs, fmt.Errorf("failed to remove %s from the registry: %w", info.ID, err)) + } + + return errors.Join(append(errs, c.Close())...) +} diff --git a/libs/migration/migration_test.go b/libs/migration/migration_test.go new file mode 100644 index 000000000..b486f376f --- /dev/null +++ b/libs/migration/migration_test.go @@ -0,0 +1,132 @@ +package migration + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/capabilities/libs/capability" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-common/pkg/types/core" +) + +// fake stands in for a generated capability server. The capability interface is embedded rather than implemented, +// because nothing here calls it. +type fake struct { + services.Service + capabilities.ExecutableAndTriggerCapability + + info capabilities.CapabilityInfo + started bool + closed bool +} + +const fakeID = "fake@1.0.0" + +func newFake(t *testing.T) *fake { + info, err := capabilities.NewCapabilityInfo(fakeID, capabilities.CapabilityTypeCombined, "a fake") + require.NoError(t, err) + + f := &fake{info: info} + f.Service, _ = services.Config{ + Name: "FakeCapability", + Start: func(context.Context) error { f.started = true; return nil }, + Close: func() error { f.closed = true; return nil }, + }.NewServiceEngine(logger.Test(t)) + return f +} + +func (f *fake) Info(context.Context) (capabilities.CapabilityInfo, error) { return f.info, nil } + +func (f *fake) Methods() []capability.Method { return nil } + +// recordingRegistry records what is added to it and removed from it. It rejects what is added when err is set. +type recordingRegistry struct { + registry.CapabilitiesRegistry + err error + added []capabilities.BaseCapability + removed []string +} + +func (r *recordingRegistry) Add(_ context.Context, c capabilities.BaseCapability) error { + if r.err != nil { + return r.err + } + + r.added = append(r.added, c) + return nil +} + +func (r *recordingRegistry) Remove(_ context.Context, id string) error { + r.removed = append(r.removed, id) + return nil +} + +func TestInitialiseServesTheCapability(t *testing.T) { + f := newFake(t) + reg := &recordingRegistry{} + deps := core.StandardCapabilitiesDependencies{CapabilityRegistry: reg, ShardIndex: 3} + + var got core.StandardCapabilitiesDependencies + s := newStandardCapabilities("fake", func(d core.StandardCapabilitiesDependencies) (capability.Capability, error) { + got = d + return f, nil + }) + + _, err := s.Infos(t.Context()) + require.ErrorIs(t, err, errNotInitialised) + require.ErrorIs(t, s.Ready(), errNotInitialised) + assert.Equal(t, map[string]error{"fake": errNotInitialised}, s.HealthReport()) + + require.NoError(t, s.Initialise(t.Context(), deps)) + assert.Equal(t, deps, got) + assert.True(t, f.started) + require.NoError(t, s.Ready()) + assert.Equal(t, []capabilities.BaseCapability{f}, reg.added) + + infos, err := s.Infos(t.Context()) + require.NoError(t, err) + assert.Equal(t, []capabilities.CapabilityInfo{f.info}, infos) + + require.NoError(t, s.Close()) + assert.True(t, f.closed) + assert.Equal(t, []string{fakeID}, reg.removed) +} + +func TestCloseBeforeInitialise(t *testing.T) { + s := newStandardCapabilities("fake", func(core.StandardCapabilitiesDependencies) (capability.Capability, error) { + t.Error("the capability must not be built") + return newFake(t), nil + }) + require.NoError(t, s.Close()) +} + +func TestInitialiseFails(t *testing.T) { + t.Run("when the constructor fails", func(t *testing.T) { + s := newStandardCapabilities("fake", func(core.StandardCapabilitiesDependencies) (capability.Capability, error) { + return nil, errors.New("no schedule") + }) + + deps := core.StandardCapabilitiesDependencies{CapabilityRegistry: &recordingRegistry{}} + require.ErrorContains(t, s.Initialise(t.Context(), deps), "failed to instantiate capability: no schedule") + }) + + t.Run("when the registry rejects the capability, closing it", func(t *testing.T) { + f := newFake(t) + s := newStandardCapabilities("fake", func(core.StandardCapabilitiesDependencies) (capability.Capability, error) { + return f, nil + }) + t.Cleanup(func() { require.NoError(t, s.Close()) }) + + deps := core.StandardCapabilitiesDependencies{CapabilityRegistry: &recordingRegistry{err: errors.New("rejected")}} + require.ErrorContains(t, s.Initialise(t.Context(), deps), "rejected") + assert.True(t, f.closed) + }) +} diff --git a/libs/migration/project.json b/libs/migration/project.json new file mode 100644 index 000000000..8c9a1baaf --- /dev/null +++ b/libs/migration/project.json @@ -0,0 +1,36 @@ +{ + "name": "migration", + "projectType": "library", + "projectRoot": "libs/migration", + "tags": [], + "targets": { + "lint": { + "executor": "nx:run-commands", + "options": { + "cwd": "{projectRoot}", + "command": "golangci-lint run ./..." + } + }, + "race": { + "executor": "nx:run-commands", + "options": { + "cwd": "{projectRoot}", + "command": "go test -race ./..." + } + }, + "test": { + "executor": "nx:run-commands", + "options": { + "cwd": "{projectRoot}", + "command": "go test ./... -coverpkg=./... -coverprofile=coverage.txt" + } + }, + "tidy": { + "executor": "nx:run-commands", + "options": { + "cwd": "{projectRoot}", + "command": "go mod tidy" + } + } + } +} diff --git a/cron/.tool-versions b/scheduler/cron/.tool-versions similarity index 100% rename from cron/.tool-versions rename to scheduler/cron/.tool-versions diff --git a/cron/README.md b/scheduler/cron/README.md similarity index 73% rename from cron/README.md rename to scheduler/cron/README.md index b3f85eb1a..0b8129f31 100644 --- a/cron/README.md +++ b/scheduler/cron/README.md @@ -16,16 +16,13 @@ A trigger that uses a cron schedule to run periodically at fixed times, dates, o ### Sample ``` -{ +payload, _ := anypb.New(&protos.Config{Schedule: "0 * * * * *"}) + +capabilities.TriggerRegistrationRequest{ TriggerID: "test-id-1", - Metadata: capabilities.RequestMetadata{ - WorkflowID: "workflow-id-1", - }, - Config: values.NewMap( - map[string]interface{}{ - "schedule": "0 * * * * *", - } - ), + Metadata: capabilities.RequestMetadata{WorkflowID: "workflow-id-1"}, + Method: "Trigger", + Payload: payload, } ``` @@ -33,20 +30,16 @@ A trigger that uses a cron schedule to run periodically at fixed times, dates, o | Name | Description | Type | Default | |:----------------------:|:------------------------------------------------------------------------:|:------:|:-------:| -| ScheduledExecutionTime | Time that cron trigger's task execution had been scheduled to occur | string | - | +| ScheduledExecutionTime | Time that cron trigger's task execution had been scheduled to occur | google.protobuf.Timestamp | - | ### Sample ``` -type Response struct { -// The ID of the trigger capability -TriggerType string -// The ID of the trigger event -ID string -// Trigger-specific payload -Outputs *values.Map -Payload: { - ScheduledExecutionTime: '2022-07-14T12:01:25.000000+08:00', -} +capabilities.TriggerEvent{ + TriggerType: "cron-trigger@1.0.0", + // The scheduled execution time, RFC3339 in UTC + ID: "2022-07-14T04:01:25Z", + // A protos.Payload + Payload: anypb.Any{ScheduledExecutionTime: 2022-07-14T04:01:25Z}, } ``` \ No newline at end of file diff --git a/cron/cron_capabilities_spec.toml b/scheduler/cron/cron_capabilities_spec.toml similarity index 100% rename from cron/cron_capabilities_spec.toml rename to scheduler/cron/cron_capabilities_spec.toml diff --git a/cron/go.mod b/scheduler/cron/go.mod similarity index 96% rename from cron/go.mod rename to scheduler/cron/go.mod index 9ee0c230d..6ea681924 100644 --- a/cron/go.mod +++ b/scheduler/cron/go.mod @@ -1,4 +1,4 @@ -module github.com/smartcontractkit/capabilities/cron +module github.com/smartcontractkit/capabilities/scheduler/cron go 1.27.3 @@ -7,8 +7,10 @@ require ( github.com/google/uuid v1.6.0 github.com/jonboulle/clockwork v0.5.0 github.com/samber/do/v2 v2.1.0 - github.com/smartcontractkit/capabilities/libs v0.0.0-20261028150945-49616d404694 + github.com/smartcontractkit/capabilities/libs v0.0.0-20261104185952-8fa7314422b5 github.com/smartcontractkit/chainlink-common v0.11.2-0.20261102201435-1c6d22d0a5ad + github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261102161601-beffc93e226e + github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b github.com/stretchr/testify v1.12.1 go.opentelemetry.io/otel v1.46.0 go.opentelemetry.io/otel/metric v1.46.0 @@ -84,8 +86,6 @@ require ( github.com/shopspring/decimal v1.4.0 // indirect github.com/smartcontractkit/chain-selectors v1.0.104 // indirect github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 // indirect - github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261014172408-e5560b3eeb33 // indirect - github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b // indirect github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b // indirect github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260710151514-27b5a126dabe // indirect github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect @@ -135,4 +135,7 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/grpc v1.83.2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect + sigs.k8s.io/yaml v1.6.0 // indirect ) + +replace github.com/smartcontractkit/capabilities/libs => ../../libs diff --git a/cron/go.sum b/scheduler/cron/go.sum similarity index 99% rename from cron/go.sum rename to scheduler/cron/go.sum index 002a4a8bb..96082dc16 100644 --- a/cron/go.sum +++ b/scheduler/cron/go.sum @@ -226,16 +226,14 @@ github.com/scylladb/go-reflectx v1.0.1 h1:b917wZM7189pZdlND9PbIJ6NQxfDPfBvUaQ7cj github.com/scylladb/go-reflectx v1.0.1/go.mod h1:rWnOfDIRWBGN0miMLIcoPt/Dhi2doCMZqwMCJ3KupFc= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= -github.com/smartcontractkit/capabilities/libs v0.0.0-20261028150945-49616d404694 h1:V9R2cz7lytOXSIutKoAncFLW2RrZWHQ/UfNcGrsdge0= -github.com/smartcontractkit/capabilities/libs v0.0.0-20261028150945-49616d404694/go.mod h1:FAL0grPNIp+lBU3s3ygVsEh89WY+5i2hjB5xzh+5Iyc= github.com/smartcontractkit/chain-selectors v1.0.104 h1:/n9pPGM5W/+r1eHoWZv4VwX9LNS1af4+ICyhM8zKRNM= github.com/smartcontractkit/chain-selectors v1.0.104/go.mod h1:qy7whtgG5g+7z0jt0nRyii9bLND9m15NZTzuQPkMZ5w= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261102201435-1c6d22d0a5ad h1:LM3VRkflY4m4PravYkX6/wEZPBw8KnYJDwjz1AFZyAs= github.com/smartcontractkit/chainlink-common v0.11.2-0.20261102201435-1c6d22d0a5ad/go.mod h1:GQPosLRkCllnWENEHTetu6W5/xuFC24QMqXm35mYmt4= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549 h1:J4vYRJxkBO8Yf//17wScKjYWmS45VZaFWx0zt+4ETEU= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20261005132539-b7d63d5fd549/go.mod h1:Scyk7O80fde5rJF3Ty3fEE3xiS6j84eWbu8/2gl3pIY= -github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261014172408-e5560b3eeb33 h1:9aS43MBP2grzsxpppcqosVkaeKFEfqfcIG2puIec3y4= -github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261014172408-e5560b3eeb33/go.mod h1:BseXIPMhfBpV2z/S+tzEFI10LbAzr7HfRIEnPQE1QQY= +github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261102161601-beffc93e226e h1:zuim5QLXqDGW8R74eWIFpfr0XmWRPi96onpxNXUSMsY= +github.com/smartcontractkit/chainlink-common/x/config v0.0.0-20261102161601-beffc93e226e/go.mod h1:D0u/3Ne/7FR3+2TYf9uMkorU2RldtQMKWcCnBghqNaw= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b h1:tlxPkVQ6ZpJIElqUWc4lU5VH4BB6OzhckG/gHwpaaa0= github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20261013212015-e622e8b83b6b/go.mod h1:/i8hjTPFdVWHiY+QjeSiVS2Z3GB3WAZznGgXHstC02E= github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b h1:QuI6SmQFK/zyUlVWEf0GMkiUYBPY4lssn26nKSd/bOM= @@ -472,3 +470,5 @@ gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.1.3/go.mod h1:NgwopIslSNH47DimFoV78dnkksY2EFtX0ajyb3K/las= +sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= +sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/scheduler/cron/legacy/main.go b/scheduler/cron/legacy/main.go new file mode 100644 index 000000000..db56e7f40 --- /dev/null +++ b/scheduler/cron/legacy/main.go @@ -0,0 +1,40 @@ +// Command legacy serves cron as a StandardCapabilities LOOP, for nodes that do not run it with capability.Run yet. +package main + +import ( + "encoding/json" + "fmt" + + "github.com/jonboulle/clockwork" + + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/libs/migration" + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + "github.com/smartcontractkit/capabilities/scheduler/cron/service" + + "github.com/smartcontractkit/chainlink-common/pkg/loop" + "github.com/smartcontractkit/chainlink-common/pkg/types/core" +) + +// serviceName is what the LOOP was named before cron moved to capability.Run. +const serviceName = "CronCapabilities" + +func main() { + migration.Serve(serviceName, newCron, loop.WithOtelViews(service.MetricViews())) +} + +func newCron(s *loop.Server, deps core.StandardCapabilitiesDependencies) (capability.Capability, error) { + var cfg service.Config + if deps.Config != "" { + if err := json.Unmarshal([]byte(deps.Config), &cfg); err != nil { + return nil, fmt.Errorf("failed to unmarshal config: %s %w", deps.Config, err) + } + } + + svc, err := service.NewService(s.Logger, clockwork.NewRealClock(), s.LimitsFactory, cfg, deps.OrgResolver, deps.ShardIndex) + if err != nil { + return nil, err + } + + return protos.NewCronServer(svc), nil +} diff --git a/scheduler/cron/main.go b/scheduler/cron/main.go new file mode 100644 index 000000000..0876b0d1f --- /dev/null +++ b/scheduler/cron/main.go @@ -0,0 +1,12 @@ +package main + +//go:generate go run . docgen + +import ( + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/scheduler/cron/service" +) + +func main() { + capability.Run("cron", service.New, capability.WithOtelViews(service.MetricViews()...)) +} diff --git a/cron/project.json b/scheduler/cron/project.json similarity index 82% rename from cron/project.json rename to scheduler/cron/project.json index df4d962e5..becf98f58 100644 --- a/cron/project.json +++ b/scheduler/cron/project.json @@ -1,7 +1,7 @@ { "name": "cron", "projectType": "capability", - "implicitDependencies": ["loopserver"], + "implicitDependencies": ["capability", "migration"], "tags": [], "targets": { "lint": { @@ -38,10 +38,10 @@ "options": { "cwd": "{projectRoot}", "commands": [ - "CGO_ENABLED='0' GOOS='linux' GOARCH='amd64' go build -o ./../bin/amd64/{projectName} .", - "CGO_ENABLED='0' GOOS='linux' GOARCH='arm64' go build -o ./../bin/arm64/{projectName} .", - "CGO_ENABLED='0' GOOS='linux' GOARCH='amd64' go build -o ./../bin/amd64/{projectName}-v2 ./cmd/cronv2", - "CGO_ENABLED='0' GOOS='linux' GOARCH='arm64' go build -o ./../bin/arm64/{projectName}-v2 ./cmd/cronv2" + "CGO_ENABLED='0' GOOS='linux' GOARCH='amd64' go build -o {workspaceRoot}/bin/amd64/{projectName} ./legacy", + "CGO_ENABLED='0' GOOS='linux' GOARCH='arm64' go build -o {workspaceRoot}/bin/arm64/{projectName} ./legacy", + "CGO_ENABLED='0' GOOS='linux' GOARCH='amd64' go build -o {workspaceRoot}/bin/amd64/{projectName}-v2 .", + "CGO_ENABLED='0' GOOS='linux' GOARCH='arm64' go build -o {workspaceRoot}/bin/arm64/{projectName}-v2 ." ] } }, @@ -51,6 +51,7 @@ "cwd": "{projectRoot}", "commands": [ "grep -rl \"^// Code generated by \"| xargs -r rm", + "go generate -x ./protos/...", "go generate -x ./..." ], "parallel": false diff --git a/scheduler/cron/protos/generate.go b/scheduler/cron/protos/generate.go new file mode 100644 index 000000000..62b0d2a31 --- /dev/null +++ b/scheduler/cron/protos/generate.go @@ -0,0 +1,2 @@ +//go:generate go run github.com/smartcontractkit/capabilities/libs/capability/cmd/protoc +package protos diff --git a/scheduler/cron/protos/production/capabilities/scheduler/cron/v1/trigger.proto b/scheduler/cron/protos/production/capabilities/scheduler/cron/v1/trigger.proto new file mode 100644 index 000000000..2d4fa7476 --- /dev/null +++ b/scheduler/cron/protos/production/capabilities/scheduler/cron/v1/trigger.proto @@ -0,0 +1,30 @@ +// Code generated by protoc-gen-cre. DO NOT EDIT. + +syntax = "proto3"; + +package capabilities.scheduler.cron.v1; + +import "google/protobuf/timestamp.proto"; + +import "tools/generator/v1alpha/cre_metadata.proto"; + +message Config { + string schedule = 1; // Cron schedule string +} + +message Payload { + google.protobuf.Timestamp scheduled_execution_time = 1; +} + +// LegacyPayload is what the retired LegacyTrigger sent, kept as it was released. +message LegacyPayload { + option deprecated = true; + + string scheduled_execution_time = 1; // Time that cron trigger's task execution had been scheduled to occur (RFC3339Nano formatted) +} + +service Cron { + option (tools.generator.v1alpha.capability) = { mode: MODE_DON, capability_id: "cron-trigger@1.0.0" }; + + rpc Trigger ( Config ) returns ( stream Payload ); +} diff --git a/scheduler/cron/protos/protostest/trigger_instances_gen.go b/scheduler/cron/protos/protostest/trigger_instances_gen.go new file mode 100644 index 000000000..5b859f402 --- /dev/null +++ b/scheduler/cron/protos/protostest/trigger_instances_gen.go @@ -0,0 +1,25 @@ +// Code generated by github.com/smartcontractkit/capabilities/libs/capability/cmd/protoc-gen-cre, DO NOT EDIT. + +package protostest + +import ( + "context" + + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + + "github.com/smartcontractkit/capabilities/libs/capability/instances" +) + +// CronInstances calls each method on every instance at once, checking that they agree. A node capability +// has one. +type CronInstances []*protos.CronServer + +func (s CronInstances) RegisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *protos.Config) (<-chan instances.TypedEvent[*protos.Payload], error) { + return instances.TypedRegisterTrigger[*protos.Payload](ctx, s, "Trigger", triggerID, metadata, input) +} + +func (s CronInstances) UnregisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *protos.Config) error { + return instances.TypedUnregisterTrigger(ctx, s, "Trigger", triggerID, metadata, input) +} diff --git a/scheduler/cron/protos/trigger.pb.go b/scheduler/cron/protos/trigger.pb.go new file mode 100644 index 000000000..a24748c26 --- /dev/null +++ b/scheduler/cron/protos/trigger.pb.go @@ -0,0 +1,231 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.12 +// protoc v5.29.3 +// source: capabilities/scheduler/cron/x/v1/trigger.proto + +// The development package: its production protos are released as capabilities.scheduler.cron.v1, which never +// changes. + +package protos + +import ( + _ "github.com/smartcontractkit/capabilities/libs/capability/feature" + _ "github.com/smartcontractkit/chainlink-protos/cre/go/tools/generator" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type Config struct { + state protoimpl.MessageState `protogen:"open.v1"` + Schedule string `protobuf:"bytes,1,opt,name=schedule,proto3" json:"schedule,omitempty"` // Cron schedule string + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Config) Reset() { + *x = Config{} + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Config) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Config) ProtoMessage() {} + +func (x *Config) ProtoReflect() protoreflect.Message { + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Config.ProtoReflect.Descriptor instead. +func (*Config) Descriptor() ([]byte, []int) { + return file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescGZIP(), []int{0} +} + +func (x *Config) GetSchedule() string { + if x != nil { + return x.Schedule + } + return "" +} + +type Payload struct { + state protoimpl.MessageState `protogen:"open.v1"` + ScheduledExecutionTime *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=scheduled_execution_time,json=scheduledExecutionTime,proto3" json:"scheduled_execution_time,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Payload) Reset() { + *x = Payload{} + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Payload) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Payload) ProtoMessage() {} + +func (x *Payload) ProtoReflect() protoreflect.Message { + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Payload.ProtoReflect.Descriptor instead. +func (*Payload) Descriptor() ([]byte, []int) { + return file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescGZIP(), []int{1} +} + +func (x *Payload) GetScheduledExecutionTime() *timestamppb.Timestamp { + if x != nil { + return x.ScheduledExecutionTime + } + return nil +} + +// LegacyPayload is what the retired LegacyTrigger sent, kept as it was released. +// +// Deprecated: Marked as deprecated in capabilities/scheduler/cron/x/v1/trigger.proto. +type LegacyPayload struct { + state protoimpl.MessageState `protogen:"open.v1"` + ScheduledExecutionTime string `protobuf:"bytes,1,opt,name=scheduled_execution_time,json=scheduledExecutionTime,proto3" json:"scheduled_execution_time,omitempty"` // Time that cron trigger's task execution had been scheduled to occur (RFC3339Nano formatted) + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LegacyPayload) Reset() { + *x = LegacyPayload{} + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LegacyPayload) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LegacyPayload) ProtoMessage() {} + +func (x *LegacyPayload) ProtoReflect() protoreflect.Message { + mi := &file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LegacyPayload.ProtoReflect.Descriptor instead. +func (*LegacyPayload) Descriptor() ([]byte, []int) { + return file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescGZIP(), []int{2} +} + +func (x *LegacyPayload) GetScheduledExecutionTime() string { + if x != nil { + return x.ScheduledExecutionTime + } + return "" +} + +var File_capabilities_scheduler_cron_x_v1_trigger_proto protoreflect.FileDescriptor + +const file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDesc = "" + + "\n" + + ".capabilities/scheduler/cron/x/v1/trigger.proto\x12 capabilities.scheduler.cron.x.v1\x1a capability/feature/feature.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a*tools/generator/v1alpha/cre_metadata.proto\"7\n" + + "\x06Config\x12-\n" + + "\bschedule\x18\x01 \x01(\tB\x11\xa2\xbb\x18\rpsb_migrationR\bschedule\"r\n" + + "\aPayload\x12g\n" + + "\x18scheduled_execution_time\x18\x01 \x01(\v2\x1a.google.protobuf.TimestampB\x11\xa2\xbb\x18\rpsb_migrationR\x16scheduledExecutionTime\"`\n" + + "\rLegacyPayload\x12K\n" + + "\x18scheduled_execution_time\x18\x01 \x01(\tB\x11\xa2\xbb\x18\rpsb_migrationR\x16scheduledExecutionTime:\x02\x18\x012\x97\x01\n" + + "\x04Cron\x12s\n" + + "\aTrigger\x12(.capabilities.scheduler.cron.x.v1.Config\x1a).capabilities.scheduler.cron.x.v1.Payload\"\x11\xaa\xbb\x18\rpsb_migration0\x01\x1a\x1a\x82\xb5\x18\x16\b\x01\x12\x12cron-trigger@1.0.0B\x11\xb2\xbb\x18\rpsb_migrationb\x06proto3" + +var ( + file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescOnce sync.Once + file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescData []byte +) + +func file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescGZIP() []byte { + file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescOnce.Do(func() { + file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDesc), len(file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDesc))) + }) + return file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDescData +} + +var file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_capabilities_scheduler_cron_x_v1_trigger_proto_goTypes = []any{ + (*Config)(nil), // 0: capabilities.scheduler.cron.x.v1.Config + (*Payload)(nil), // 1: capabilities.scheduler.cron.x.v1.Payload + (*LegacyPayload)(nil), // 2: capabilities.scheduler.cron.x.v1.LegacyPayload + (*timestamppb.Timestamp)(nil), // 3: google.protobuf.Timestamp +} +var file_capabilities_scheduler_cron_x_v1_trigger_proto_depIdxs = []int32{ + 3, // 0: capabilities.scheduler.cron.x.v1.Payload.scheduled_execution_time:type_name -> google.protobuf.Timestamp + 0, // 1: capabilities.scheduler.cron.x.v1.Cron.Trigger:input_type -> capabilities.scheduler.cron.x.v1.Config + 1, // 2: capabilities.scheduler.cron.x.v1.Cron.Trigger:output_type -> capabilities.scheduler.cron.x.v1.Payload + 2, // [2:3] is the sub-list for method output_type + 1, // [1:2] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_capabilities_scheduler_cron_x_v1_trigger_proto_init() } +func file_capabilities_scheduler_cron_x_v1_trigger_proto_init() { + if File_capabilities_scheduler_cron_x_v1_trigger_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDesc), len(file_capabilities_scheduler_cron_x_v1_trigger_proto_rawDesc)), + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_capabilities_scheduler_cron_x_v1_trigger_proto_goTypes, + DependencyIndexes: file_capabilities_scheduler_cron_x_v1_trigger_proto_depIdxs, + MessageInfos: file_capabilities_scheduler_cron_x_v1_trigger_proto_msgTypes, + }.Build() + File_capabilities_scheduler_cron_x_v1_trigger_proto = out.File + file_capabilities_scheduler_cron_x_v1_trigger_proto_goTypes = nil + file_capabilities_scheduler_cron_x_v1_trigger_proto_depIdxs = nil +} diff --git a/scheduler/cron/protos/trigger.proto b/scheduler/cron/protos/trigger.proto new file mode 100644 index 000000000..981b16c96 --- /dev/null +++ b/scheduler/cron/protos/trigger.proto @@ -0,0 +1,36 @@ +syntax = "proto3"; + +// The development package: its production protos are released as capabilities.scheduler.cron.v1, which never +// changes. +package capabilities.scheduler.cron.x.v1; + +import "capability/feature/feature.proto"; +import "google/protobuf/timestamp.proto"; +import "tools/generator/v1alpha/cre_metadata.proto"; + +option (cre.feature.v1.released) = "psb_migration"; + +message Config { + string schedule = 1 [(cre.feature.v1.field) = "psb_migration"]; // Cron schedule string +} + +message Payload { + google.protobuf.Timestamp scheduled_execution_time = 1 [(cre.feature.v1.field) = "psb_migration"]; +} + +// LegacyPayload is what the retired LegacyTrigger sent, kept as it was released. +message LegacyPayload { + option deprecated = true; + string scheduled_execution_time = 1 [(cre.feature.v1.field) = "psb_migration"]; // Time that cron trigger's task execution had been scheduled to occur (RFC3339Nano formatted) +} + +service Cron { + option (tools.generator.v1alpha.capability) = { + mode: MODE_DON + capability_id: "cron-trigger@1.0.0" + }; + + rpc Trigger(Config) returns (stream Payload) { + option (cre.feature.v1.method) = "psb_migration"; + } +} diff --git a/scheduler/cron/protos/trigger_server_gen.go b/scheduler/cron/protos/trigger_server_gen.go new file mode 100644 index 000000000..aa87f50ab --- /dev/null +++ b/scheduler/cron/protos/trigger_server_gen.go @@ -0,0 +1,130 @@ +// Code generated by github.com/smartcontractkit/capabilities/libs/capability/cmd/protoc-gen-cre, DO NOT EDIT. + +package protos + +import ( + "context" + "fmt" + "sync" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" + + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/libs/capability/feature" + "github.com/smartcontractkit/capabilities/libs/capability/server" +) + +// Requests are read with feature.UnmarshalAny, so a capability built from its development protos serves requests +// made with its production protos too, and replies with the kind of message the request was. + +type CronCapability interface { + RegisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *Config) (<-chan capabilities.TriggerAndId[*Payload], caperrors.Error) + UnregisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *Config) caperrors.Error + AckEvent(ctx context.Context, triggerId string, eventId string, method string) caperrors.Error + + Start(ctx context.Context) error + Close() error + HealthReport() map[string]error + Name() string + Description() string + Ready() error +} + +func NewCronServer(capability CronCapability) *CronServer { + stopCh := make(chan struct{}) + return &CronServer{ + cronCapability: cronCapability{CronCapability: capability, stopCh: stopCh}, + stopCh: stopCh, + } +} + +// CronServer serves a CronCapability as a capability.Capability. +type CronServer struct { + cronCapability + stopCh chan struct{} + stop sync.Once +} + +// Close stops answering registered triggers, then closes the capability. +func (c *CronServer) Close() error { + c.stop.Do(func() { close(c.stopCh) }) + + return c.cronCapability.Close() +} + +type cronCapability struct { + CronCapability + stopCh chan struct{} +} + +func (c *cronCapability) Info(ctx context.Context) (capabilities.CapabilityInfo, error) { + return capabilities.NewCapabilityInfo("cron-trigger@1.0.0", capabilities.CapabilityTypeCombined, c.CronCapability.Description()) +} + +var _ capabilities.ExecutableAndTriggerCapability = (*cronCapability)(nil) + +const CronID = "cron-trigger@1.0.0" + +func (c *cronCapability) Methods() []capability.Method { + return []capability.Method{ + {Name: "Trigger", Trigger: true, Input: &Config{}, Output: &Payload{}}, + } +} + +func (c *cronCapability) RegisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) { + ctx = request.Metadata.ContextWithCRE(ctx) + switch request.Method { + case "Trigger": + input := &Config{} + production, err := feature.UnmarshalAny(request.Payload, input) + if err != nil { + return nil, fmt.Errorf("failed to read the trigger's config: %w", err) + } + events, caperr := c.CronCapability.RegisterTrigger(ctx, request.TriggerID, request.Metadata, input) + if caperr != nil { + return nil, caperr + } + return server.ForwardTriggers(c.stopCh, "cron-trigger@1.0.0", production, events), nil + default: + return nil, fmt.Errorf("trigger %s not found", request.Method) + } +} + +func (c *cronCapability) UnregisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) error { + ctx = request.Metadata.ContextWithCRE(ctx) + switch request.Method { + case "Trigger": + input := &Config{} + if _, err := feature.UnmarshalAny(request.Payload, input); err != nil { + return fmt.Errorf("failed to read the trigger's config: %w", err) + } + if err := c.CronCapability.UnregisterTrigger(ctx, request.TriggerID, request.Metadata, input); err != nil { + return err + } + return nil + default: + return fmt.Errorf("trigger %s not found", request.Method) + } +} + +func (c *cronCapability) AckEvent(ctx context.Context, triggerId string, eventId string, method string) error { + switch method { + case "Trigger": + return c.CronCapability.AckEvent(ctx, triggerId, eventId, method) + default: + return fmt.Errorf("trigger %s not found", method) + } +} + +func (c *cronCapability) RegisterToWorkflow(ctx context.Context, request capabilities.RegisterToWorkflowRequest) error { + return nil +} + +func (c *cronCapability) UnregisterFromWorkflow(ctx context.Context, request capabilities.UnregisterFromWorkflowRequest) error { + return nil +} + +func (c *cronCapability) Execute(ctx context.Context, request capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) { + return capabilities.CapabilityResponse{}, fmt.Errorf("method %s not found", request.Method) +} diff --git a/scheduler/cron/service/capability.go b/scheduler/cron/service/capability.go new file mode 100644 index 000000000..2b7c6e00e --- /dev/null +++ b/scheduler/cron/service/capability.go @@ -0,0 +1,56 @@ +package service + +import ( + "github.com/jonboulle/clockwork" + "github.com/samber/do/v2" + + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" +) + +// ConfigNamespace is the namespace the cron capability's Config is set under. +const ConfigNamespace = "cron" + +// init provides Config however the capability is run, so whoever embeds it gets its defaults without providing it. +func init() { + capability.Injectors.MustProvideConfig(ConfigNamespace, &Config{}) +} + +var _ capability.Capability = (*protos.CronServer)(nil) + +// New builds the cron capability from what the run provides, for capability.Run. An OrgResolver is +// optional; without one, organization IDs are not fetched. +func New(i do.Injector) (*protos.CronServer, error) { + lggr, err := do.Invoke[logger.Logger](i) + if err != nil { + return nil, err + } + + clock, err := do.Invoke[clockwork.Clock](i) + if err != nil { + return nil, err + } + + limitsFactory, err := do.Invoke[limits.Factory](i) + if err != nil { + return nil, err + } + + cfg, err := do.Invoke[*Config](i) + if err != nil { + return nil, err + } + + orgResolver, _ := do.Invoke[orgresolver.OrgResolver](i) + capCfg, _ := do.Invoke[capability.CapabilitiesConfig](i) + svc, err := NewService(lggr, clock, limitsFactory, *cfg, orgResolver, capCfg.ShardIndex) + if err != nil { + return nil, err + } + + return protos.NewCronServer(svc), nil +} diff --git a/scheduler/cron/service/capability_test.go b/scheduler/cron/service/capability_test.go new file mode 100644 index 000000000..2c0a820d5 --- /dev/null +++ b/scheduler/cron/service/capability_test.go @@ -0,0 +1,20 @@ +package service + +import ( + "testing" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" +) + +func TestNew(t *testing.T) { + c := startCrons(t, clockwork.NewFakeClock(), Config{FastestScheduleIntervalSeconds: 1}, 1)[0] + + info, err := c.Info(t.Context()) + require.NoError(t, err) + assert.Equal(t, protos.CronID, info.ID) + assert.NoError(t, c.Ready()) +} diff --git a/scheduler/cron/service/doccomments_gen.go b/scheduler/cron/service/doccomments_gen.go new file mode 100644 index 000000000..649eafb8e --- /dev/null +++ b/scheduler/cron/service/doccomments_gen.go @@ -0,0 +1,11 @@ +// Code generated by github.com/smartcontractkit/capabilities/scheduler/cron docgen, DO NOT EDIT. + +package service + +import "github.com/smartcontractkit/chainlink-common/x/config/commentparsing" + +func (Config) DocComments() map[string]commentparsing.FieldDoc { + return map[string]commentparsing.FieldDoc{ + "FastestScheduleIntervalSeconds": {Comment: "FastestScheduleIntervalSeconds is the shortest interval a schedule may run at. Unset uses the\nCRE settings default."}, + } +} diff --git a/cron/trigger/logger.go b/scheduler/cron/service/logger.go similarity index 98% rename from cron/trigger/logger.go rename to scheduler/cron/service/logger.go index 90e3b236e..1a95e6908 100644 --- a/cron/trigger/logger.go +++ b/scheduler/cron/service/logger.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "sync" diff --git a/cron/trigger/metrics.go b/scheduler/cron/service/metrics.go similarity index 99% rename from cron/trigger/metrics.go rename to scheduler/cron/service/metrics.go index 62de8f7a1..af3b4fb97 100644 --- a/cron/trigger/metrics.go +++ b/scheduler/cron/service/metrics.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "context" diff --git a/cron/trigger/monitor.go b/scheduler/cron/service/monitor.go similarity index 97% rename from cron/trigger/monitor.go rename to scheduler/cron/service/monitor.go index fb4a55a49..6dfeaa652 100644 --- a/cron/trigger/monitor.go +++ b/scheduler/cron/service/monitor.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "context" diff --git a/cron/trigger/trigger.go b/scheduler/cron/service/service.go similarity index 86% rename from cron/trigger/trigger.go rename to scheduler/cron/service/service.go index 4c8c2f7fd..430bf6991 100644 --- a/cron/trigger/trigger.go +++ b/scheduler/cron/service/service.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "context" @@ -13,33 +13,30 @@ import ( "github.com/jonboulle/clockwork" "google.golang.org/protobuf/types/known/timestamppb" + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" - crontypedapi "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron" - "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/triggers/cron/server" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" - "github.com/smartcontractkit/chainlink-common/pkg/types/core" "github.com/smartcontractkit/chainlink-common/pkg/workflows" "github.com/smartcontractkit/chainlink-common/pkg/workflows/events" ) -const ServiceName = "CronCapabilities" - const defaultSendChannelBufferSize = 1000 var cronTriggerInfo = capabilities.MustNewCapabilityInfo( - server.CronID, + protos.CronID, capabilities.CapabilityTypeTrigger, "A trigger that uses a cron schedule to run periodically at fixed times, dates, or intervals.", ) type Config struct { - // FastestScheduleIntervalSeconds is the shortest interval a cron schedule may have. Zero keeps the + // FastestScheduleIntervalSeconds is the shortest interval a schedule may run at. Unset uses the // CRE settings default. FastestScheduleIntervalSeconds int `json:"fastestScheduleIntervalSeconds"` } @@ -63,26 +60,10 @@ type cronTrigger struct { close func() } -// TODO: Remove when the generated server.CronCapability no longer requires -// these methods. -type unimplemented struct{} - -func (u unimplemented) Initialise(context.Context, core.StandardCapabilitiesDependencies) error { - return errors.ErrUnsupported -} -func (u unimplemented) RegisterLegacyTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *crontypedapi.Config) (<-chan capabilities.TriggerAndId[*crontypedapi.LegacyPayload], caperrors.Error) { //nolint:staticcheck - return nil, caperrors.NewPublicSystemError(errors.ErrUnsupported, caperrors.Unimplemented) -} - -func (u unimplemented) UnregisterLegacyTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *crontypedapi.Config) caperrors.Error { - return caperrors.NewPublicSystemError(errors.ErrUnsupported, caperrors.Unimplemented) -} - +// Service is the cron trigger, served by protos.NewCronServer. type Service struct { - unimplemented services.StateMachine capabilities.CapabilityInfo - limitsFactory limits.Factory fastestScheduleInterval limits.TimeLimiter clock clockwork.Clock lggr logger.Logger @@ -97,9 +78,10 @@ type Service struct { var _ services.Service = &Service{} -// NewService creates a new trigger service. Optionally, a clock can be passed in for testing, if nil -// the system clock will be used. -func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory limits.Factory, cfg Config) (_ *Service, err error) { +// NewService creates a new trigger service, scheduling on clock, for the node shard shardIndex. The orgResolver is +// optional and can be nil, but should be set in live environments. +func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory limits.Factory, cfg Config, orgResolver orgresolver.OrgResolver, + shardIndex uint32) (_ *Service, err error) { lggr := logger.Named(parentLggr, "CRONTrigger") fastestScheduleInterval, err := newFastestScheduleInterval(limitsFactory, cfg) @@ -117,6 +99,10 @@ func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory l return nil, fmt.Errorf("error creating metrics: %w", err) } + if orgResolver == nil { + lggr.Warn("OrgResolver is nil, cron capability will not be able to fetch organization ID") + } + var options []gocron.SchedulerOption options = append(options, gocron.WithMonitor(NewCronMonitor(metrics))) // Set scheduler location to UTC for consistency across nodes. @@ -124,12 +110,7 @@ func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory l // Adapt chainlink logger to gocron logger interface. cronLogger := NewCronLogger(lggr) options = append(options, gocron.WithLogger(cronLogger)) - // Allow injecting a clock for testing. Otherwise use system clock. - if clock != nil { - options = append(options, gocron.WithClock(clock)) - } else { - clock = clockwork.NewRealClock() - } + options = append(options, gocron.WithClock(clock)) scheduler, err := gocron.NewScheduler(options...) if err != nil { @@ -139,7 +120,6 @@ func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory l return &Service{ lggr: lggr, CapabilityInfo: cronTriggerInfo, - limitsFactory: limitsFactory, fastestScheduleInterval: fastestScheduleInterval, triggers: NewCronStore(), scheduler: scheduler, @@ -150,11 +130,13 @@ func NewService(parentLggr logger.Logger, clock clockwork.Clock, limitsFactory l "capabilityVersion", cronTriggerInfo.Version(), "capabilityName", cronTriggerInfo.ID, ), - metrics: metrics, + metrics: metrics, + orgResolver: orgResolver, + shardIndex: shardIndex, }, nil } -func (s *Service) RegisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *crontypedapi.Config) (<-chan capabilities.TriggerAndId[*crontypedapi.Payload], caperrors.Error) { +func (s *Service) RegisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *protos.Config) (<-chan capabilities.TriggerAndId[*protos.Payload], caperrors.Error) { ctx = metadata.ContextWithCRE(ctx) var muCh sync.RWMutex // extra synchronization to prevent the cron task from racing to send on the closed chan and re-register itself // hold the lock until we call triggers.Write @@ -167,7 +149,7 @@ func (s *Service) RegisterTrigger(ctx context.Context, triggerID string, metadat } var job gocron.Job - callbackCh := make(chan capabilities.TriggerAndId[*crontypedapi.Payload], defaultSendChannelBufferSize) + callbackCh := make(chan capabilities.TriggerAndId[*protos.Payload], defaultSendChannelBufferSize) closeCh := func() { muCh.Lock() @@ -344,7 +326,7 @@ func (s *Service) RegisterTrigger(ctx context.Context, triggerID string, metadat return callbackCh, nil } -func createTriggerResponse(scheduledExecutionTime time.Time) capabilities.TriggerAndId[*crontypedapi.Payload] { +func createTriggerResponse(scheduledExecutionTime time.Time) capabilities.TriggerAndId[*protos.Payload] { // Ensure UTC time is used for consistency across nodes. scheduledExecutionTimeUTC := scheduledExecutionTime.UTC() @@ -354,8 +336,8 @@ func createTriggerResponse(scheduledExecutionTime time.Time) capabilities.Trigge scheduledExecutionTimeFormatted := scheduledExecutionTimeUTC.Format(time.RFC3339) triggerEventID := scheduledExecutionTimeFormatted - return capabilities.TriggerAndId[*crontypedapi.Payload]{ - Trigger: &crontypedapi.Payload{ + return capabilities.TriggerAndId[*protos.Payload]{ + Trigger: &protos.Payload{ ScheduledExecutionTime: timestamppb.New(scheduledExecutionTimeUTC), }, Id: triggerEventID, @@ -366,7 +348,7 @@ func (s *Service) AckEvent(ctx context.Context, triggerID string, eventID string return nil } -func (s *Service) UnregisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *crontypedapi.Config) caperrors.Error { +func (s *Service) UnregisterTrigger(ctx context.Context, triggerID string, metadata capabilities.RequestMetadata, input *protos.Config) caperrors.Error { trigger, ok := s.triggers.Read(triggerID) if !ok { s.lggr.Warnw("trigger not found", "triggerID", triggerID) @@ -423,13 +405,16 @@ func (s *Service) start() error { // Close stops the Service. // After this call the Service cannot be started again, -// The service will need to be re-built to start scheduling again. +// The service will need to be re-built to start scheduling again. Closing it again does nothing. func (s *Service) Close() error { err := s.StopOnce(s.Name(), s.close) - if errors.Is(err, services.ErrCannotStopUnstarted) { + switch { + case errors.Is(err, services.ErrCannotStopUnstarted): // gocron runs a goroutine from construction rather than from Start, so a service closed // without having started, e.g. when Initialise fails, still has to shut the scheduler down. return s.close() + case errors.Is(err, services.ErrAlreadyStopped): + return nil } return err } diff --git a/scheduler/cron/service/service_test.go b/scheduler/cron/service/service_test.go new file mode 100644 index 000000000..7ee69313e --- /dev/null +++ b/scheduler/cron/service/service_test.go @@ -0,0 +1,311 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/go-co-op/gocron/v2" + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + "github.com/smartcontractkit/chainlink-common/pkg/workflows" +) + +// These tests need the Service itself rather than what a run builds. + +func TestCronTrigger_CloseStartErrors(t *testing.T) { + fakeClock := clockwork.NewRealClock() + ts, err := NewService(logger.Nop(), fakeClock, limits.Factory{}, Config{}, nil, 0) + require.NoError(t, err) + ctx := t.Context() + + err = ts.Start(ctx) + require.NoError(t, err) + err = ts.Close() + require.NoError(t, err) + err = ts.Start(ctx) + require.Error(t, err) +} + +func TestCronTrigger_ReadyAndHealthFollowLifecycle(t *testing.T) { + ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, Config{}, nil, 0) + require.NoError(t, err) + + require.Error(t, ts.Ready()) + require.Error(t, ts.HealthReport()[ts.Name()]) + + require.NoError(t, ts.Start(t.Context())) + require.NoError(t, ts.Ready()) + require.NoError(t, ts.HealthReport()[ts.Name()]) + + require.NoError(t, ts.Close()) + require.Error(t, ts.Ready()) + require.Error(t, ts.HealthReport()[ts.Name()]) +} + +func TestCronTrigger_CloseBeforeStart(t *testing.T) { + ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, Config{}, nil, 0) + require.NoError(t, err) + + require.NoError(t, ts.Close()) + require.NoError(t, ts.Close()) +} + +func TestCronTrigger_CloseTwice(t *testing.T) { + ts, err := NewService(logger.Nop(), clockwork.NewRealClock(), limits.Factory{}, Config{}, nil, 0) + require.NoError(t, err) + require.NoError(t, ts.Start(t.Context())) + + require.NoError(t, ts.Close()) + require.NoError(t, ts.Close()) +} + +// TestCronTrigger_NoLogsAfterClose covers gocron, which keeps logging for a moment after Shutdown returns: +// none of it may reach the service's logger once Close has returned. +func TestCronTrigger_NoLogsAfterClose(t *testing.T) { + for range 20 { + lggr, observedLogs := logger.TestObserved(t, zap.DebugLevel) + ts, err := NewService(lggr, clockwork.NewFakeClock(), limits.Factory{}, Config{}, nil, 0) + require.NoError(t, err) + require.NoError(t, ts.Start(t.Context())) + + require.NoError(t, ts.Close()) + closed := observedLogs.Len() + + time.Sleep(10 * time.Millisecond) + require.Equal(t, closed, observedLogs.Len(), "logged after Close returned: %v", observedLogs.All()[closed:]) + } +} + +type panicOnNowClock struct { + clockwork.Clock +} + +func (c *panicOnNowClock) Now() time.Time { + panic("clock panic in Now()") +} + +func TestGocronNewTaskPanic(t *testing.T) { + fakeClock := clockwork.NewFakeClock() + + lggr, observedLogs := logger.TestObserved(t, zap.ErrorLevel) + ts := newStartedService(t, lggr, fakeClock, Config{FastestScheduleIntervalSeconds: 1}) + + _, capErr := ts.RegisterTrigger(t.Context(), triggerID1, capabilities.RequestMetadata{WorkflowID: workflowID1}, &protos.Config{Schedule: everySecond}) + require.Nil(t, capErr) + + // Swap ts.clock after setup so gocron's scheduler/executor keep using the + // original fakeClock, while the task callback (trigger.go s.clock.Now()) + // hits the panicking clock. + ts.clock = &panicOnNowClock{Clock: fakeClock} + + fakeClock.Advance(time.Second * 1) + + ticker := time.NewTicker(time.Second * 1) + timeout := time.NewTimer(time.Second * 10) + for { + select { + case <-ticker.C: + logs := observedLogs.All() + for _, log := range logs { + if log.Message == "panic in gocron.NewTask function" && len(log.Context) > 0 && log.Context[0].Key == "err" && log.Context[0].String == "clock panic in Now()" { + return + } + } + case <-timeout.C: + t.Fatal("timeout, no panic in gocron.NewTask function") + } + } +} + +// MockOrgResolver is a mock implementation of the OrgResolver interface +type MockOrgResolver struct { + mock.Mock +} + +func (m *MockOrgResolver) Get(ctx context.Context, owner string) (string, error) { + args := m.Called(ctx, owner) + return args.String(0), args.Error(1) +} + +func (m *MockOrgResolver) Start(ctx context.Context) error { + return nil +} + +func (m *MockOrgResolver) Close() error { + return nil +} + +func (m *MockOrgResolver) HealthReport() map[string]error { + return map[string]error{} +} + +func (m *MockOrgResolver) Ready() error { + return nil +} + +func (m *MockOrgResolver) Name() string { + return "MockOrgResolver" +} + +// Ensure MockOrgResolver implements the interface +var _ orgresolver.OrgResolver = (*MockOrgResolver)(nil) + +// TestCronTrigger_ExecutionIDWithTriggerIndex verifies execution IDs include the +// trigger index parsed from ReferenceID. +func TestCronTrigger_ExecutionIDWithTriggerIndex(t *testing.T) { + t.Parallel() + + const testWorkflowID = "workflow-id-multi-trigger-test" + const testReferenceID = "trigger_2" // parsed to triggerIndex = 2 + const testTriggerIndex = 2 + const testTriggerID = testWorkflowID + "|test-trigger" + + startTime := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + fakeClock := clockwork.NewFakeClockAt(startTime) + lggr, observedLogs := logger.TestObserved(t, zap.DebugLevel) + + ts := newStartedService(t, lggr, fakeClock, Config{FastestScheduleIntervalSeconds: 1}) + + metadata := capabilities.RequestMetadata{ + WorkflowID: testWorkflowID, + ReferenceID: testReferenceID, + } + ch, capErr := ts.RegisterTrigger(t.Context(), testTriggerID, metadata, &protos.Config{Schedule: everySecond}) + require.Nil(t, capErr) + + fakeClock.Advance(time.Second + time.Millisecond) + msg := <-ch + + expectedExecID, err := workflows.GenerateExecutionIDWithTriggerIndex(testWorkflowID, msg.Id, testTriggerIndex) + require.NoError(t, err) + + // The debug log at "sending trigger event" is written before the channel + // send, so it is already present once we receive msg. + var execIDFromLog string + var found bool + for _, entry := range observedLogs.All() { + if entry.Message == "sending trigger event" { + for _, field := range entry.Context { + if field.Key == "executionID" { + execIDFromLog = field.String + } + } + found = true + break + } + } + require.True(t, found, "expected log entry 'sending trigger event'") + require.Equal(t, expectedExecID, execIDFromLog, "execution ID should match expected hash function") + + require.Nil(t, ts.UnregisterTrigger(t.Context(), testTriggerID, metadata, &protos.Config{Schedule: everySecond})) +} + +func TestEnforceFastestSchedule_NonUniformSecondsField(t *testing.T) { + t.Parallel() + // a schedule that has a bunch of 5s gaps followed by a bunch of 1s gaps + // requires checking multiple runs to catch the short gaps + schedule := "0,5,10,15,20,25,30,35,40,45,50,55-59 * * * * *" + maximumFastest := 5 * time.Second + + jobDef := gocron.CronJob(schedule, true) + 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()) +} + +// newStartedService is a started service on clock, closed when the test ends if the test has not +// closed it itself. +func newStartedService(t *testing.T, lggr logger.Logger, clock clockwork.Clock, cfg Config) *Service { + t.Helper() + + ts, err := NewService(lggr, clock, limits.Factory{}, cfg, nil, 0) + require.NoError(t, err) + require.NoError(t, ts.Start(t.Context())) + t.Cleanup(func() { _ = ts.Close() }) + + return ts +} diff --git a/cron/trigger/store.go b/scheduler/cron/service/store.go similarity index 98% rename from cron/trigger/store.go rename to scheduler/cron/service/store.go index 9484c2970..dacf41c35 100644 --- a/cron/trigger/store.go +++ b/scheduler/cron/service/store.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "maps" diff --git a/scheduler/cron/service/trigger_test.go b/scheduler/cron/service/trigger_test.go new file mode 100644 index 000000000..49aab47a4 --- /dev/null +++ b/scheduler/cron/service/trigger_test.go @@ -0,0 +1,439 @@ +package service + +import ( + "errors" + "strconv" + "sync" + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/capabilities/libs/capability" + "github.com/smartcontractkit/capabilities/libs/capability/capabilitytest" + "github.com/smartcontractkit/capabilities/libs/capability/instances" + "github.com/smartcontractkit/capabilities/scheduler/cron/protos" + "github.com/smartcontractkit/capabilities/scheduler/cron/protos/protostest" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + caperrors "github.com/smartcontractkit/chainlink-common/pkg/capabilities/errors" +) + +const ( + // Schedules + everyYear = "0 0 0 1 1 *" + everyMonth = "0 0 0 1 * *" + everyWeek = "0 0 0 * * 0" + everyDay = "0 0 0 * * *" + everyDayEasternTZ = "TZ=America/New_York 0 0 * * * *" + everyHour = "0 0 * * * *" + everyHourFrom9To10 = "0 9-10 * * *" + everyMinute = "0 * * * * *" + everySecond = "* * * * * *" + everyEvenSecond = "*/2 * * * * *" + + // Workflow IDs + workflowID1 = "workflow-id-1" + + // Trigger IDs + triggerID1 = "test-id-1" + triggerID2 = "test-id-2" + + // instanceCount is how many instances most tests run, so every event is also checked for agreement. + instanceCount = 4 +) + +var metadata = capabilities.RequestMetadata{WorkflowID: workflowID1} + +// startCrons starts n instances of the cron trigger as capability.Run would, all scheduling on clock. +func startCrons(t *testing.T, clock clockwork.Clock, cfg Config, n int) protostest.CronInstances { + t.Helper() + + return capabilitytest.Start(t, New, n, capability.WithProvider(func(i *capability.Injector) error { + return errors.Join(i.ProvideValue(clock), i.ProvideValue(&cfg)) + })) +} + +// trigger is one registered trigger and the events its instances agreed on. +type trigger struct { + id string + input *protos.Config + events <-chan instances.TypedEvent[*protos.Payload] +} + +func register(t *testing.T, crons protostest.CronInstances, id, schedule string) *trigger { + t.Helper() + + input := &protos.Config{Schedule: schedule} + events, err := crons.RegisterTrigger(t.Context(), id, metadata, input) + require.NoError(t, err) + return &trigger{id: id, input: input, events: events} +} + +func (tr *trigger) unregister(t *testing.T, crons protostest.CronInstances) { + t.Helper() + require.NoError(t, crons.UnregisterTrigger(t.Context(), tr.id, metadata, tr.input)) +} + +// next is the next event every instance agreed on. +func (tr *trigger) next(t *testing.T) capabilities.TriggerAndId[*protos.Payload] { + t.Helper() + + select { + case e, ok := <-tr.events: + require.True(t, ok, "the trigger's events ended") + require.Nil(t, e.Mismatch) + return e.TriggerAndId + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for an event") + return capabilities.TriggerAndId[*protos.Payload]{} + } +} + +// scheduled is the time the next event was scheduled for. +func (tr *trigger) scheduled(t *testing.T) time.Time { + t.Helper() + return tr.next(t).Trigger.ScheduledExecutionTime.AsTime() +} + +// requireNone fails if an event arrives soon. +func (tr *trigger) requireNone(t *testing.T) { + t.Helper() + + select { + case e := <-tr.events: + t.Fatalf("unexpected event %v", e) + case <-time.After(100 * time.Millisecond): + } +} + +// requireClosed fails unless the trigger's events end. +func (tr *trigger) requireClosed(t *testing.T) { + t.Helper() + + select { + case _, ok := <-tr.events: + require.False(t, ok, "the trigger's events should have ended") + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for the trigger's events to end") + } +} + +func TestCronTrigger_SuccessWithStandardCronIntervals(t *testing.T) { + cases := []struct { + name string + schedule string + interval [4]time.Duration + }{ + {name: "every second", schedule: everySecond, interval: [4]time.Duration{time.Second, time.Second, time.Second, time.Second}}, + {name: "every minute", schedule: everyMinute, interval: [4]time.Duration{time.Minute, time.Minute, time.Minute, time.Minute}}, + {name: "every hour", schedule: everyHour, interval: [4]time.Duration{time.Hour, time.Hour, time.Hour, time.Hour}}, + {name: "every day", schedule: everyDay, interval: [4]time.Duration{24 * time.Hour, 24 * time.Hour, 24 * time.Hour, 24 * time.Hour}}, + {name: "every week on Sunday", schedule: everyWeek, interval: [4]time.Duration{7 * 24 * time.Hour, 7 * 24 * time.Hour, 7 * 24 * time.Hour, 7 * 24 * time.Hour}}, + {name: "every month", schedule: everyMonth, interval: [4]time.Duration{31 * 24 * time.Hour, 28 * 24 * time.Hour, 31 * 24 * time.Hour, 30 * 24 * time.Hour}}, + {name: "every year", schedule: everyYear, interval: [4]time.Duration{365 * 24 * time.Hour, 365 * 24 * time.Hour, 365 * 24 * time.Hour, 365 * 24 * time.Hour}}, + } + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + // Year 1 starts on a Monday and has no leap years for the intervals above to step over; the + // weekly case starts on the Sunday after. + clock := clockwork.NewFakeClockAt(time.Time{}) + if tt.schedule == everyWeek { + clock.Advance(6 * 24 * time.Hour) + } + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, tt.schedule) + + clock.Advance(tt.interval[0] - time.Millisecond) + tr.requireNone(t) + clock.Advance(2 * time.Millisecond) + first := tr.scheduled(t) + + clock.Advance(tt.interval[1]) + second := tr.scheduled(t) + + clock.Advance(tt.interval[2]) + third := tr.scheduled(t) + + tr.unregister(t, crons) + clock.Advance(tt.interval[3]) + tr.requireClosed(t) + + assert.Equal(t, first.Add(tt.interval[1]), second) + assert.Equal(t, second.Add(tt.interval[2]), third) + }) + } +} + +// TestCronTrigger_Deterministic is what a DON relies on: every instance fires on the same tick with the +// same event, tick after tick. +func TestCronTrigger_Deterministic(t *testing.T) { + const ticks = 100 + + start := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + clock := clockwork.NewFakeClockAt(start) + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, everySecond) + + for i := 1; i <= ticks; i++ { + clock.Advance(time.Second) + e := tr.next(t) + + want := start.Add(time.Duration(i) * time.Second) + require.Equal(t, want, e.Trigger.ScheduledExecutionTime.AsTime(), "tick %d", i) + require.Equal(t, want.Format(time.RFC3339), e.Id, "tick %d", i) + } + + tr.unregister(t, crons) + tr.requireClosed(t) +} + +// TestCronTrigger_Load runs on the real clock, and on one instance since it is the scheduler under load +// being measured rather than agreement. +func TestCronTrigger_Load(t *testing.T) { + t.Parallel() + const numTriggers = 1_000 + const numExecutions = 3 + + clock := clockwork.NewRealClock() + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, 1) + + triggers := make([]*trigger, numTriggers) + for i := range triggers { + triggers[i] = register(t, crons, "test-id-"+strconv.Itoa(i+1), everySecond) + } + + var ( + mu sync.Mutex + scheduled [numTriggers][numExecutions]time.Time + actual [numTriggers][numExecutions]time.Time + wg sync.WaitGroup + ) + for i, tr := range triggers { + wg.Go(func() { + for exec := range numExecutions { + s := tr.scheduled(t) + now := clock.Now().UTC() + + mu.Lock() + scheduled[i][exec] = s + actual[i][exec] = now + mu.Unlock() + } + }) + } + wg.Wait() + + for _, tr := range triggers { + tr.unregister(t, crons) + } + for _, tr := range triggers { + tr.requireClosed(t) + } + + mu.Lock() + defer mu.Unlock() + var totalDelta time.Duration + for exec := range numExecutions { + earliest, latest := scheduled[0][exec], scheduled[0][exec] + for i := range numTriggers { + s, a := scheduled[i][exec], actual[i][exec] + if s.Before(earliest) { + earliest = s + } + if s.After(latest) { + latest = s + } + if exec > 0 { + require.Equal(t, scheduled[i][exec-1].Add(time.Second), s, "executions should be a second apart") + } + require.False(t, a.Before(s), "an execution should not run before it is scheduled") + require.False(t, a.After(s.Add(time.Second)), "an execution should run within a second of when it is scheduled") + totalDelta += a.Sub(s) + } + // Triggers on the same schedule stay aligned, though registration on the real clock can span a + // second boundary under load. + require.False(t, latest.After(earliest.Add(time.Second))) + } + t.Logf("Average delta: %s", totalDelta/(numTriggers*numExecutions)) +} + +func TestCronTrigger_RegisterTriggerBeforeStart(t *testing.T) { + clock := clockwork.NewRealClock() + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, everySecond) + + first := tr.scheduled(t) + firstAt := clock.Now().UTC() + second := tr.scheduled(t) + secondAt := clock.Now().UTC() + + tr.unregister(t, crons) + + assert.Equal(t, first.Add(time.Second), second) + for _, at := range []struct{ scheduled, actual time.Time }{{first, firstAt}, {second, secondAt}} { + assert.True(t, at.actual.After(at.scheduled), "an execution should not run before it is scheduled") + assert.False(t, at.actual.After(at.scheduled.Add(time.Second)), "an execution should run within a second of when it is scheduled") + } +} + +func TestCronTriggerTimeWindows(t *testing.T) { + // 08:50 UTC + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 8, 50, 0, 0, time.UTC)) + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, everyHourFrom9To10) + + clock.Advance(10*time.Minute + time.Millisecond) + nine := tr.scheduled(t) + + clock.Advance(time.Hour) + ten := tr.scheduled(t) + + // Not again until 09:00 the next day. + clock.Advance(23 * time.Hour) + nextNine := tr.scheduled(t) + + tr.unregister(t, crons) + + assert.Equal(t, nine.Add(time.Hour), ten) + assert.Equal(t, nine.Add(24*time.Hour), nextNine) +} + +func TestCronTrigger_MultipleDifferentSchedules(t *testing.T) { + // An odd second, so the even schedule first fires a second after the every-second one. + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 0, 0, 1, 0, time.UTC)) + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + every := register(t, crons, triggerID1, everySecond) + even := register(t, crons, triggerID2, everyEvenSecond) + + var everyEvents, evenEvents []capabilities.TriggerAndId[*protos.Payload] + for second := 1; second <= 4; second++ { + clock.Advance(time.Second) + everyEvents = append(everyEvents, every.next(t)) + if second%2 == 1 { + evenEvents = append(evenEvents, even.next(t)) + } + } + + every.unregister(t, crons) + even.unregister(t, crons) + clock.Advance(time.Second) + every.requireClosed(t) + even.requireClosed(t) + + for i := 1; i < len(everyEvents); i++ { + assert.Equal(t, everyEvents[i-1].Trigger.ScheduledExecutionTime.AsTime().Add(time.Second), everyEvents[i].Trigger.ScheduledExecutionTime.AsTime()) + } + assert.Equal(t, evenEvents[0].Trigger.ScheduledExecutionTime.AsTime().Add(2*time.Second), evenEvents[1].Trigger.ScheduledExecutionTime.AsTime()) + // The even schedule's events land on the every-second schedule's 1st and 3rd, and share their IDs. + assert.Equal(t, everyEvents[0].Id, evenEvents[0].Id) + assert.Equal(t, everyEvents[2].Id, evenEvents[1].Id) +} + +func TestCronTriggerTimeZone(t *testing.T) { + eastern, err := time.LoadLocation("America/New_York") + require.NoError(t, err) + + // 23:50 Eastern + clock := clockwork.NewFakeClockAt(time.Date(2024, 1, 1, 23, 50, 0, 0, eastern)) + crons := startCrons(t, clock, Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, everyDayEasternTZ) + + clock.Advance(10*time.Minute - time.Millisecond) + tr.requireNone(t) + + clock.Advance(time.Millisecond) + midnight := tr.scheduled(t) + + tr.unregister(t, crons) + + assert.True(t, midnight.Equal(time.Date(2024, 1, 2, 0, 0, 0, 0, eastern)), "fired at %s", midnight) +} + +func TestCronTrigger_RegisterTrigger(t *testing.T) { + cases := []struct { + name string + schedule string + expectedErr string + code caperrors.ErrorCode + }{ + {name: "valid cron schedule - 6 entries", schedule: everyMinute}, + {name: "valid cron schedule - 5 entries", schedule: "* * * * *"}, + { + name: "invalid cron schedule - empty", + expectedErr: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nexpected 5 to 6 fields, found 0: []", + }, + { + name: "invalid cron schedule - not a cron schedule", + schedule: "d d d d d", + expectedErr: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nfailed to parse int from d: strconv.Atoi: parsing \"d\": invalid syntax", + }, + { + name: "invalid cron schedule - invalid timezone", + schedule: "TZ=moon * * * * *", + expectedErr: "[3]InvalidArgument: failed to initialize job: gocron: CronJob: crontab parse failure\nprovided bad location moon: unknown time zone moon", + }, + { + name: "invalid cron schedule - exceeds maximum fastest", + schedule: everySecond, + expectedErr: "[101]LimitExceeded: maximum fastest cron schedule is 30s", + }, + } + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + crons := startCrons(t, clockwork.NewRealClock(), Config{}, instanceCount) + + _, err := crons.RegisterTrigger(t.Context(), triggerID1, metadata, &protos.Config{Schedule: tt.schedule}) + if tt.expectedErr == "" { + require.NoError(t, err) + return + } + + require.EqualError(t, err, tt.expectedErr) + var capErr caperrors.Error + require.ErrorAs(t, err, &capErr) + assert.Equal(t, caperrors.OriginUser, capErr.Origin()) + assert.Equal(t, caperrors.VisibilityPublic, capErr.Visibility()) + }) + } +} + +func TestCronTrigger_RegisterTriggerDuplicateError(t *testing.T) { + crons := startCrons(t, clockwork.NewRealClock(), Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + register(t, crons, triggerID1, everySecond) + + _, err := crons.RegisterTrigger(t.Context(), triggerID1, metadata, &protos.Config{Schedule: everySecond}) + require.EqualError(t, err, "[13]Internal: triggerId test-id-1 already registered") +} + +func TestCronTrigger_UnregisterTriggerError(t *testing.T) { + input := &protos.Config{Schedule: everySecond} + + t.Run("OK if trigger not found", func(t *testing.T) { + crons := startCrons(t, clockwork.NewRealClock(), Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + require.NoError(t, crons.UnregisterTrigger(t.Context(), "missing", metadata, input)) + }) + + t.Run("OK register then unregister multiple times", func(t *testing.T) { + crons := startCrons(t, clockwork.NewRealClock(), Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + tr := register(t, crons, triggerID1, everySecond) + for range 3 { + tr.unregister(t, crons) + } + }) + + t.Run("NOK fails to unregister if closed", func(t *testing.T) { + crons := startCrons(t, clockwork.NewRealClock(), Config{FastestScheduleIntervalSeconds: 1}, instanceCount) + register(t, crons, triggerID1, everySecond) + for _, c := range crons { + require.NoError(t, c.Close()) + } + + require.ErrorContains(t, crons.UnregisterTrigger(t.Context(), triggerID1, metadata, input), + "cannot unregister a new trigger, service has been closed") + }) +} diff --git a/cron/trigger/utils.go b/scheduler/cron/service/utils.go similarity index 99% rename from cron/trigger/utils.go rename to scheduler/cron/service/utils.go index 382ad5763..e76e16c4a 100644 --- a/cron/trigger/utils.go +++ b/scheduler/cron/service/utils.go @@ -1,4 +1,4 @@ -package trigger +package service import ( "fmt"