diff --git a/core/capabilities/testutils/backend.go b/core/capabilities/testutils/backend.go index 6efa3048758..f41bcc6838d 100644 --- a/core/capabilities/testutils/backend.go +++ b/core/capabilities/testutils/backend.go @@ -103,7 +103,7 @@ func (th *EVMBackendTH) SetupCoreServices(t *testing.T) (logpoller.HeadTracker, return ht, lp } -func (th *EVMBackendTH) NewContractReader(t *testing.T, ctx context.Context, cfg []byte) (types.ContractReader, error) { +func (th *EVMBackendTH) NewContractReader(ctx context.Context, t *testing.T, cfg []byte) (types.ContractReader, error) { crCfg := &evmrelaytypes.ChainReaderConfig{} if err := json.Unmarshal(cfg, crCfg); err != nil { return nil, err diff --git a/core/capabilities/testutils/chain_reader.go b/core/capabilities/testutils/chain_reader.go index cf9e0ea5c71..976099c2a73 100644 --- a/core/capabilities/testutils/chain_reader.go +++ b/core/capabilities/testutils/chain_reader.go @@ -8,11 +8,11 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/test-go/testify/require" + commoncaps "github.com/smartcontractkit/chainlink-common/pkg/capabilities" commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" commonvalues "github.com/smartcontractkit/chainlink-common/pkg/values" - "github.com/test-go/testify/require" - "github.com/smartcontractkit/chainlink/v2/core/capabilities/triggers/logevent" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/log_emitter" coretestutils "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" @@ -94,7 +94,7 @@ func NewContractReaderTH(t *testing.T) *ContractReaderTH { // Create a new contract reader to return from mock relayer ctx := coretestutils.Context(t) - contractReader, err := backendTH.NewContractReader(t, ctx, contractReaderCfgBytes) + contractReader, err := backendTH.NewContractReader(ctx, t, contractReaderCfgBytes) require.NoError(t, err) return &ContractReaderTH{ diff --git a/core/capabilities/triggers/logevent/service.go b/core/capabilities/triggers/logevent/service.go index 399c40fac8d..36b6b467d53 100644 --- a/core/capabilities/triggers/logevent/service.go +++ b/core/capabilities/triggers/logevent/service.go @@ -22,12 +22,14 @@ type Input struct { // Log Event Trigger Capabilities Manager // Manages different log event triggers using an underlying triggerStore type TriggerService struct { + services.StateMachine capabilities.CapabilityInfo capabilities.Validator[RequestConfig, Input, capabilities.TriggerResponse] lggr logger.Logger triggers CapabilitiesStore[logEventTrigger, capabilities.TriggerResponse] relayer core.Relayer logEventConfig Config + stopCh services.StopChan } // Common capability level config across all workflows @@ -63,6 +65,7 @@ func NewTriggerService(ctx context.Context, p Params) (*TriggerService, error) { triggers: logEventStore, relayer: p.Relayer, logEventConfig: p.LogEventConfig, + stopCh: make(services.StopChan), } var err error s.CapabilityInfo, err = s.Info(ctx) @@ -93,14 +96,19 @@ func (s *TriggerService) RegisterTrigger(ctx context.Context, req capabilities.T } // Add log event trigger with Contract details to CapabilitiesStore var respCh chan capabilities.TriggerResponse - respCh, err = s.triggers.InsertIfNotExists(req.TriggerID, func() (*logEventTrigger, chan capabilities.TriggerResponse, error) { - l, ch, tErr := newLogEventTrigger(ctx, s.lggr, req.Metadata.WorkflowID, reqConfig, s.logEventConfig, s.relayer) - if tErr != nil { + ok := s.IfNotStopped(func() { + respCh, err = s.triggers.InsertIfNotExists(req.TriggerID, func() (*logEventTrigger, chan capabilities.TriggerResponse, error) { + l, ch, tErr := newLogEventTrigger(ctx, s.lggr, req.Metadata.WorkflowID, reqConfig, s.logEventConfig, s.relayer) + if tErr != nil { + return l, ch, tErr + } + tErr = l.Start(ctx) return l, ch, tErr - } - tErr = l.Start(ctx) - return l, ch, tErr + }) }) + if !ok { + return nil, fmt.Errorf("cannot create new trigger since LogEventTriggerService has been stopped") + } if err != nil { return nil, fmt.Errorf("create new trigger failed %w", err) } @@ -133,8 +141,11 @@ func (s *TriggerService) Start(ctx context.Context) error { // After this call the Service cannot be started again, // The service will need to be re-built to start scheduling again. func (s *TriggerService) Close() error { - triggers := s.triggers.ReadAll() - return services.MultiCloser(triggers).Close() + return s.StopOnce("Log Event Trigger Capability Service", func() error { + s.lggr.Infow("Stopping LogEventTrigger Capability Service") + triggers := s.triggers.ReadAll() + return services.MultiCloser(triggers).Close() + }) } func (s *TriggerService) Ready() error {