diff --git a/core/services/workflows/syncer/v2/shard_failover_manager.go b/core/services/workflows/syncer/v2/shard_failover_manager.go index 265792de952..987f359816e 100644 --- a/core/services/workflows/syncer/v2/shard_failover_manager.go +++ b/core/services/workflows/syncer/v2/shard_failover_manager.go @@ -19,6 +19,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/workflows/shardownership" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/store" v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" ) // cachedExpiry is how long a cached trigger event is kept for potential failover replay. @@ -53,7 +54,7 @@ type ShardFailoverManager struct { } type cachedEvent struct { - event v2.RoutedTriggerEvent + event triggers.CoordinatedEvent cachedAt time.Time } @@ -140,7 +141,7 @@ func (m *ShardFailoverManager) close() error { // admissionCheck is wired as the engine's OnTriggerAdmission hook. It // checks shard ownership dynamically per-trigger and decides whether the // engine should process the event. -func (m *ShardFailoverManager) admissionCheck(ctx context.Context, event v2.RoutedTriggerEvent) error { +func (m *ShardFailoverManager) admissionCheck(ctx context.Context, event triggers.CoordinatedEvent) error { if !m.cfg.ShardingEnabled { return nil } @@ -252,7 +253,7 @@ func (m *ShardFailoverManager) checkShardOwnership(ctx context.Context) shardown } } -func (m *ShardFailoverManager) cacheEvent(event v2.RoutedTriggerEvent) { +func (m *ShardFailoverManager) cacheEvent(event triggers.CoordinatedEvent) { eventID := event.Event.Event.ID m.mu.Lock() m.cache[eventID] = cachedEvent{event: event, cachedAt: time.Now()} @@ -264,12 +265,12 @@ func (m *ShardFailoverManager) cacheEvent(event v2.RoutedTriggerEvent) { "triggerIndex", event.TriggerIndex) } -func (m *ShardFailoverManager) popCachedEvent(eventID string) (v2.RoutedTriggerEvent, bool) { +func (m *ShardFailoverManager) popCachedEvent(eventID string) (triggers.CoordinatedEvent, bool) { m.mu.Lock() defer m.mu.Unlock() ce, ok := m.cache[eventID] if !ok { - return v2.RoutedTriggerEvent{}, false + return triggers.CoordinatedEvent{}, false } delete(m.cache, eventID) return ce.event, true diff --git a/core/services/workflows/v2/base_engine.go b/core/services/workflows/v2/base_engine.go index f30bf49934c..d5879177a70 100644 --- a/core/services/workflows/v2/base_engine.go +++ b/core/services/workflows/v2/base_engine.go @@ -41,6 +41,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/workflows/monitoring" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/store" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" "github.com/smartcontractkit/chainlink/v2/core/utils/safe" ) @@ -185,7 +186,7 @@ func (e *baseEngine) initServiceEngine(lggr logger.SugaredLogger, name string, s } // ExecuteTrigger is the engine's single execution entry point. It performs no admission control, the caller is responsible for those. -func (e *baseEngine) ExecuteTrigger(ctx context.Context, event RoutedTriggerEvent) error { +func (e *baseEngine) ExecuteTrigger(ctx context.Context, event triggers.CoordinatedEvent) error { e.activeExecutions.Add(1) defer e.activeExecutions.Add(-1) @@ -414,7 +415,7 @@ func (e *baseEngine) localNodeSync(ctx context.Context) { } // startExecution initiates a new workflow execution, blocking until completed -func (e *baseEngine) startExecution(ctx context.Context, event RoutedTriggerEvent) error { +func (e *baseEngine) startExecution(ctx context.Context, event triggers.CoordinatedEvent) error { triggerDrop := func(reason string) { e.metrics.With(platform.KeyTriggerID, event.TriggerCapID).IncrementTriggerEventDroppedTotal(ctx, reason) } @@ -458,7 +459,7 @@ func (e *baseEngine) startExecution(ctx context.Context, event RoutedTriggerEven tm.IncrementTriggerExecutionDeduplicatedCounter(ctx) tm.IncrementWorkflowTriggerEventErrorCounter(ctx) tm.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonDuplicateExecution) - registrationID := TriggerRegistrationID(e.cfg.WorkflowID, event.TriggerIndex) + registrationID := triggers.RegistrationID(e.cfg.WorkflowID, event.TriggerIndex) ackErr := e.cfg.TriggerAcknowledger.Ack(ctx, event.TriggerCapID, registrationID, triggerEvent.ID) if ackErr != nil { e.lggr.Errorw("failed to re-ACK trigger event", "eventID", triggerEvent.ID, "err", ackErr) @@ -574,7 +575,7 @@ func (e *baseEngine) startExecution(ctx context.Context, event RoutedTriggerEven } _ = events.EmitExecutionStartedEvent(ctx, loggerLabels, triggerEvent.ID, executionID) - registrationID := TriggerRegistrationID(e.cfg.WorkflowID, event.TriggerIndex) + registrationID := triggers.RegistrationID(e.cfg.WorkflowID, event.TriggerIndex) err = e.cfg.TriggerAcknowledger.Ack(ctx, event.TriggerCapID, registrationID, triggerEvent.ID) if err != nil { e.lggr.Errorf("failed to ACK trigger event (eventID=%s): %v", triggerEvent.ID, err) @@ -1048,7 +1049,3 @@ func resolveOrgID(ctx context.Context, resolver orgresolver.OrgResolver, workflo } return resolvedOrg{ID: orgID} } - -func TriggerRegistrationID(workflowID string, triggerIndex int) string { - return fmt.Sprintf("trigger_reg_%s_%d", workflowID, triggerIndex) -} diff --git a/core/services/workflows/v2/config.go b/core/services/workflows/v2/config.go index bfeecbe1153..268c994c56e 100644 --- a/core/services/workflows/v2/config.go +++ b/core/services/workflows/v2/config.go @@ -30,6 +30,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/workflows/shardownership" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/store" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" ) type EngineConfig struct { @@ -92,7 +93,7 @@ type EngineLimiters struct { TriggerSubscriptionTime limits.TimeLimiter TriggerRegistrationsTime limits.TimeLimiter TriggerSubscription limits.BoundLimiter[int] - TriggerEventQueue limits.QueueLimiter[RoutedTriggerEvent] + TriggerEventQueue limits.QueueLimiter[triggers.CoordinatedEvent] TriggerEventQueueTimeout limits.BoundLimiter[time.Duration] ExecutionConcurrency limits.ResourcePoolLimiter[int] @@ -158,7 +159,7 @@ func (l *EngineLimiters) init(lf limits.Factory, cfgFn func(*cresettings.Workflo if err != nil { return err } - l.TriggerEventQueue, err = limits.MakeQueueLimiter[RoutedTriggerEvent](lf, cfg.TriggerEventQueueLimit) + l.TriggerEventQueue, err = limits.MakeQueueLimiter[triggers.CoordinatedEvent](lf, cfg.TriggerEventQueueLimit) if err != nil { return err } @@ -442,7 +443,7 @@ type LifecycleHooks struct { // - ErrAdmissionCache: the event was cached by the admission layer; // the engine drops it without ACKing. // - any other error: the event is denied; the engine ACKs and drops it. - OnTriggerAdmission func(ctx context.Context, event RoutedTriggerEvent) error + OnTriggerAdmission func(ctx context.Context, event triggers.CoordinatedEvent) error // Used by the standalone engine OnRequirementsSet func(executionId string, requirements *sdkpb.Requirements) @@ -537,7 +538,7 @@ func (h *LifecycleHooks) setDefaultHooks() { } } if h.OnTriggerAdmission == nil { - h.OnTriggerAdmission = func(_ context.Context, _ RoutedTriggerEvent) error { return nil } + h.OnTriggerAdmission = func(_ context.Context, _ triggers.CoordinatedEvent) error { return nil } } if h.OnRateLimited == nil { h.OnRateLimited = func(executionID string) {} diff --git a/core/services/workflows/v2/coordinated_engine.go b/core/services/workflows/v2/coordinated_engine.go index fefa34130cf..d4079ae13a5 100644 --- a/core/services/workflows/v2/coordinated_engine.go +++ b/core/services/workflows/v2/coordinated_engine.go @@ -10,6 +10,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/contexts" sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" ) var _ WorkflowEngine = (*coordinatedEngine)(nil) @@ -62,7 +63,7 @@ func (e *coordinatedEngine) Name() string { return e.base.Name() } -func (e *coordinatedEngine) ExecuteTrigger(ctx context.Context, event RoutedTriggerEvent) error { +func (e *coordinatedEngine) ExecuteTrigger(ctx context.Context, event triggers.CoordinatedEvent) error { return e.base.ExecuteTrigger(ctx, event) } diff --git a/core/services/workflows/v2/coordinated_engine_test.go b/core/services/workflows/v2/coordinated_engine_test.go index 68ed5816214..22a2998aeda 100644 --- a/core/services/workflows/v2/coordinated_engine_test.go +++ b/core/services/workflows/v2/coordinated_engine_test.go @@ -14,6 +14,7 @@ import ( modulemocks "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/mocks" sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" "github.com/smartcontractkit/chainlink/v2/core/utils/matches" ) @@ -58,7 +59,7 @@ func TestCoordinatedEngine_ExecuteTrigger(t *testing.T) { Owner: baseCfg.WorkflowOwner, Workflow: baseCfg.WorkflowID, }) - event := v2.RoutedTriggerEvent{ + event := triggers.CoordinatedEvent{ WorkflowID: baseCfg.WorkflowID, TriggerCapID: "id_0", TriggerIndex: 0, diff --git a/core/services/workflows/v2/engine.go b/core/services/workflows/v2/engine.go index 43c9ff2ed7a..4b1d4a33990 100644 --- a/core/services/workflows/v2/engine.go +++ b/core/services/workflows/v2/engine.go @@ -3,18 +3,13 @@ package v2 import ( "context" "errors" - "fmt" "sync" "sync/atomic" "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" - "golang.org/x/sync/errgroup" - "google.golang.org/protobuf/encoding/protojson" - "google.golang.org/protobuf/types/known/anypb" - "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/contexts" "github.com/smartcontractkit/chainlink-common/pkg/settings" "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" @@ -23,11 +18,14 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/platform" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/monitoring" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" ) -var _ Acknowledger = (*engine)(nil) -var _ EventSink = (*engine)(nil) -var _ WorkflowEngine = (*engine)(nil) +var ( + _ Acknowledger = (*engine)(nil) + _ EventSink = (*engine)(nil) + _ WorkflowEngine = (*engine)(nil) +) // engine is the legacy trigger-owning workflow engine: it wraps the shared // execution machinery and adds the extra responsibilities of trigger registration, @@ -40,21 +38,15 @@ type engine struct { workflowLimitUsed atomic.Bool // true if GlobalWorkflowLimit must be freed - // registration ID -> trigger capability - triggers map[string]*triggerCapability + // registration ID -> trigger handle + triggers map[string]*triggers.Handle // used to separate registration and unregistration phases triggersRegMu sync.Mutex - allTriggerEventsQueueCh limits.QueueLimiter[RoutedTriggerEvent] + allTriggerEventsQueueCh limits.QueueLimiter[triggers.CoordinatedEvent] executionsSemaphore limits.ResourcePoolLimiter[int] } -type triggerCapability struct { - capabilities.TriggerCapability - payload *anypb.Any - method string -} - // NewEngine constructs the legacy trigger-owning engine: it registers its own // triggers, holds the handles, acknowledges through itself, and owns the // workflow-count limit. @@ -66,7 +58,7 @@ func NewEngine(cfg *EngineConfig) (WorkflowEngine, error) { e := &engine{ base: base, - triggers: make(map[string]*triggerCapability), + triggers: make(map[string]*triggers.Handle), allTriggerEventsQueueCh: cfg.LocalLimiters.TriggerEventQueue, executionsSemaphore: cfg.LocalLimiters.ExecutionConcurrency, } @@ -101,7 +93,7 @@ func (e *engine) Name() string { return e.base.Name() } -func (e *engine) ExecuteTrigger(ctx context.Context, event RoutedTriggerEvent) error { +func (e *engine) ExecuteTrigger(ctx context.Context, event triggers.CoordinatedEvent) error { return e.base.ExecuteTrigger(ctx, event) } @@ -214,230 +206,75 @@ func (e *engine) close() error { } func (e *engine) runTriggerSubscriptionPhase(ctx context.Context, subscriptions []*sdkpb.TriggerSubscription) error { - // check if all requested triggers exist in the registry - triggers := make([]capabilities.TriggerCapability, 0, len(subscriptions)) - for _, sub := range subscriptions { - _, labels, _ := capabilities.ParseID(sub.Id) - chainSelector, err2 := capabilities.ChainSelectorLabel(labels) - if err2 != nil { - return fmt.Errorf("invalid chain selector for ID %s: %w", sub.Id, err2) - } - if chainSelector != nil { - err2 := e.base.cfg.LocalLimiters.ChainAllowed.AllowErr(contexts.WithChainSelector(ctx, *chainSelector)) - if err2 != nil { - if errors.Is(err2, limits.ErrorNotAllowed{}) { - return fmt.Errorf("unable to subscribe to capability %s: ChainSelector %d: %w", sub.Id, *chainSelector, err2) - } - return fmt.Errorf("failed to check access for ChainSelector %d: %w", *chainSelector, err2) - } - } - triggerCap, triggerErr := e.base.cfg.CapRegistry.GetTrigger(ctx, sub.Id) - if triggerErr != nil { - return fmt.Errorf("trigger capability not found: %w", triggerErr) - } - triggers = append(triggers, triggerCap) + var creGetter settings.Getter + if e.base.cfg.LocalLimiters != nil { + creGetter = e.base.cfg.LocalLimiters.Settings } - // register to all triggers concurrently - regCtx, regCancel, err := e.base.cfg.LocalLimiters.TriggerRegistrationsTime.WithTimeout(ctx) - if err != nil { - return err + deps := triggers.RegisterDeps{ + CapRegistry: e.base.cfg.CapRegistry, + RegTimeout: e.base.cfg.LocalLimiters.TriggerRegistrationsTime, + ChainAllowed: e.base.cfg.LocalLimiters.ChainAllowed, + Settings: creGetter, + Logger: e.base.logger(), + Metrics: e.base.metrics, } - defer regCancel() - - // trigger registration results for use in concurrent trigger subscriptions - type triggerRegResult struct { - index int - registrationID string - triggerCap capabilities.TriggerCapability - eventCh <-chan capabilities.TriggerResponse - payload *anypb.Any - method string - triggerCapID string + meta := triggers.RegisterMetadata{ + WorkflowID: e.base.cfg.WorkflowID, + WorkflowOwner: e.base.cfg.WorkflowOwner, + WorkflowName: e.base.cfg.WorkflowName, + WorkflowTag: e.base.cfg.WorkflowTag, + WorkflowDonID: e.base.localNode.Load().WorkflowDON.ID, + WorkflowDonConfigVersion: pinnedWorkflowDonConfigVersion, + WorkflowRegistryChainSelector: e.base.cfg.WorkflowRegistryChainSelector, + WorkflowRegistryAddress: e.base.cfg.WorkflowRegistryAddress, + OrgID: e.base.orgID, } - resultsCh := make(chan triggerRegResult, len(subscriptions)) - g, gCtx := errgroup.WithContext(regCtx) - - // Launch concurrent trigger registrations - for i, sub := range subscriptions { - triggerCap := triggers[i] - g.Go(func() error { - registrationID := TriggerRegistrationID(e.base.cfg.WorkflowID, i) - args := []any{"triggerID", sub.Id, "method", sub.Method} - if sub.Payload != nil { - args = append(args, "payload", protojson.Format(sub.Payload)) - } - e.base.logger().Infow("Registering trigger", args...) - metadata := capabilities.RequestMetadata{ - WorkflowID: e.base.cfg.WorkflowID, - WorkflowOwner: e.base.cfg.WorkflowOwner, - WorkflowName: e.base.cfg.WorkflowName.Hex(), - WorkflowTag: e.base.cfg.WorkflowTag, - DecodedWorkflowName: e.base.cfg.WorkflowName.String(), - WorkflowDonID: e.base.localNode.Load().WorkflowDON.ID, - WorkflowDonConfigVersion: pinnedWorkflowDonConfigVersion, - ReferenceID: fmt.Sprintf("trigger_%d", i), - WorkflowRegistryChainSelector: e.base.cfg.WorkflowRegistryChainSelector, - WorkflowRegistryAddress: e.base.cfg.WorkflowRegistryAddress, - EngineVersion: platform.ValueWorkflowVersionV2, - // no WorkflowExecutionID needed (or available at this stage) - } - var creGetter settings.Getter - if e.base.cfg.LocalLimiters != nil { - creGetter = e.base.cfg.LocalLimiters.Settings - } - propagateOrgIDMeta, _ := cresettings.Default.PropagateOrgIDInRequestMetadata.GetOrDefault(gCtx, creGetter) - if propagateOrgIDMeta && e.base.orgID != "" { - metadata.OrgID = e.base.orgID - } - triggerEventCh, regErr := triggerCap.RegisterTrigger(gCtx, capabilities.TriggerRegistrationRequest{ - TriggerID: registrationID, - Metadata: metadata, - Payload: sub.Payload, - Method: sub.Method, - // no Config needed - NoDAG uses Payload - }) - if regErr != nil { - e.base.logger().Errorw("Trigger registration failed", "triggerID", sub.Id, "err", regErr) - e.base.metrics.With(platform.KeyTriggerID, sub.Id).IncrementRegisterTriggerFailureCounter(gCtx) - return fmt.Errorf("failed to register trigger %s: %w", sub.Id, regErr) - } - // Send successful result - resultsCh <- triggerRegResult{ - index: i, - registrationID: registrationID, - triggerCap: triggerCap, - eventCh: triggerEventCh, - payload: sub.Payload, - method: sub.Method, - triggerCapID: sub.Id, - } - return nil - }) + triggerCapIDs, handles, eventChans, err := triggers.Register(ctx, deps, meta, subscriptions) + if err != nil { + return err } - // wait for all registrations to complete. - // returns first non-nil error. - registrationErr := g.Wait() - close(resultsCh) - - // Collect results into e.triggers map e.triggersRegMu.Lock() - defer e.triggersRegMu.Unlock() - - eventChans := make([]<-chan capabilities.TriggerResponse, len(subscriptions)) - triggerCapIDs := make([]string, len(subscriptions)) - - for result := range resultsCh { - e.triggers[result.registrationID] = &triggerCapability{ - TriggerCapability: result.triggerCap, - payload: result.payload, - method: result.method, - } - eventChans[result.index] = result.eventCh - triggerCapIDs[result.index] = result.triggerCapID - } - - // If any registration failed, unregister successful ones and return error - if registrationErr != nil { - e.base.logger().Errorw("One or more trigger registrations failed - reverting all", "err", registrationErr) - e.unregisterAllTriggers(ctx) // needs to be called under e.triggersRegMu lock - return registrationErr - } + e.triggers = handles + e.triggersRegMu.Unlock() - // start listening for trigger events only if all registrations succeeded + // start listening for trigger events now that all registrations succeeded for idx, triggerEventCh := range eventChans { - e.base.srvcEng.GoCtx(context.WithoutCancel(ctx), func(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case event, isOpen := <-triggerEventCh: - if !isOpen { - return - } - triggerID := subscriptions[idx].Id - eventID := event.Event.ID - e.base.metrics.With(platform.KeyTriggerID, triggerID).IncrementTriggerEventReceivedCounter(ctx) - e.base.logger().Debugw("Processing trigger event", "triggerID", triggerID, "eventID", eventID) - if event.Err != nil { - e.base.logger().Errorw("Received a trigger event with error, dropping", "triggerID", triggerID, "err", event.Err) - tm := e.base.metrics.With(platform.KeyTriggerID, triggerID) - tm.IncrementWorkflowTriggerEventErrorCounter(ctx) - tm.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonTriggerResponseError) - continue - } - - routed := RoutedTriggerEvent{ - WorkflowID: e.base.cfg.WorkflowID, - TriggerCapID: triggerID, - TriggerIndex: idx, - ObservedAt: e.base.cfg.Clock.Now(), - SequenceNumber: 0, - Event: event, - } - - if err := e.put(ctx, routed); err != nil { - // Draining is expected during workflow deletion, so it logs at info rather than error level. - if errors.Is(err, ErrEngineDraining) { - e.base.logger().Infow("Dropping trigger event: engine draining", "triggerID", triggerID, "eventID", eventID) - } else { - e.base.logger().Errorw("Failed to put routed trigger event", "triggerID", triggerID, "eventID", eventID, "err", err) - } - } + triggerID := subscriptions[idx].Id + deliver := func(ctx context.Context, routed triggers.CoordinatedEvent) { + eventID := routed.Event.Event.ID + if err := e.put(ctx, routed); err != nil { + // Draining is expected during workflow deletion, so it logs at info rather than error level. + if errors.Is(err, ErrEngineDraining) { + e.base.logger().Infow("Dropping trigger event: engine draining", "triggerID", triggerID, "eventID", eventID) + } else { + e.base.logger().Errorw("Failed to put routed trigger event", "triggerID", triggerID, "eventID", eventID, "err", err) } } + } + e.base.srvcEng.GoCtx(context.WithoutCancel(ctx), func(ctx context.Context) { + triggers.ReadLoop(ctx, e.base.logger(), e.base.metrics, e.base.cfg.Clock, e.base.cfg.WorkflowID, triggerID, idx, triggerEventCh, deliver) }) } - e.base.logger().Infow("All triggers registered successfully", "numTriggers", len(subscriptions), "triggerIDs", triggerCapIDs) - e.base.metrics.IncrementWorkflowRegisteredCounter(ctx) e.base.cfg.Hooks.OnSubscribedToTriggers(triggerCapIDs) return nil } // NOTE: needs to be called under the triggersRegMu lock func (e *engine) unregisterAllTriggers(ctx context.Context) { - failCount := 0 - for registrationID, trigger := range e.triggers { - err := trigger.UnregisterTrigger(ctx, capabilities.TriggerRegistrationRequest{ - TriggerID: registrationID, - Metadata: capabilities.RequestMetadata{ - WorkflowID: e.base.cfg.WorkflowID, - WorkflowDonID: e.base.localNode.Load().WorkflowDON.ID, - }, - Payload: trigger.payload, - Method: trigger.method, - }) - if err != nil { - e.base.logger().Errorw("Failed to unregister trigger", "registrationId", registrationID, "err", err) - failCount++ - } - } + failCount := triggers.Unregister(ctx, e.base.logger(), e.base.cfg.WorkflowID, e.base.localNode.Load().WorkflowDON.ID, e.triggers) e.base.logger().Infow("All triggers unregistered", "numTriggers", len(e.triggers), "failed", failCount) - e.triggers = make(map[string]*triggerCapability) + e.triggers = make(map[string]*triggers.Handle) } func (e *engine) Ack(ctx context.Context, triggerCapID, triggerRegistrationID, eventID string) error { - e.base.logger().Infow("ACKing trigger event", "triggerRegistrationID", triggerRegistrationID, "eventID", eventID) - - tm := e.base.metrics.With(platform.KeyTriggerID, triggerCapID) - e.triggersRegMu.Lock() - trigger, ok := e.triggers[triggerRegistrationID] + handle := e.triggers[triggerRegistrationID] e.triggersRegMu.Unlock() - if !ok { - tm.IncrementTriggerEventAckFailureCounter(ctx) - return fmt.Errorf("failed to find trigger %s", triggerRegistrationID) - } - err := trigger.AckEvent(ctx, triggerRegistrationID, eventID, trigger.method) - if err != nil { - tm.IncrementTriggerEventAckFailureCounter(ctx) - return err - } - tm.IncrementTriggerEventAckSuccessCounter(ctx) - return nil + return triggers.Ack(ctx, e.base.logger(), e.base.metrics, triggerCapID, triggerRegistrationID, eventID, handle) } // errObservedAtMissing guards put's deadline derivation: put computes the queue @@ -446,7 +283,7 @@ func (e *engine) Ack(ctx context.Context, triggerCapID, triggerRegistrationID, e var errObservedAtMissing = errors.New("trigger event ObservedAt not set") // put enqueues a trigger event into the engine's internal queue. -func (e *engine) put(ctx context.Context, event RoutedTriggerEvent) error { +func (e *engine) put(ctx context.Context, event triggers.CoordinatedEvent) error { triggerID := event.TriggerCapID eventID := event.Event.Event.ID idx := event.TriggerIndex @@ -468,7 +305,7 @@ func (e *engine) put(ctx context.Context, event RoutedTriggerEvent) error { return err } // Denied: ACK and drop - registrationID := TriggerRegistrationID(e.base.cfg.WorkflowID, event.TriggerIndex) + registrationID := triggers.RegistrationID(e.base.cfg.WorkflowID, event.TriggerIndex) if ackErr := e.base.cfg.TriggerAcknowledger.Ack(ctx, event.TriggerCapID, registrationID, eventID); ackErr != nil { e.base.logger().Errorw("failed to ACK trigger after admission denial", "eventID", eventID, "err", ackErr) } diff --git a/core/services/workflows/v2/engine_drop_paths_test.go b/core/services/workflows/v2/engine_drop_paths_test.go index 28a83d1262e..10f1d5b03b9 100644 --- a/core/services/workflows/v2/engine_drop_paths_test.go +++ b/core/services/workflows/v2/engine_drop_paths_test.go @@ -29,6 +29,7 @@ import ( workflowEvents "github.com/smartcontractkit/chainlink/v2/core/services/workflows/events" metmocks "github.com/smartcontractkit/chainlink/v2/core/services/workflows/metering/mocks" v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" "github.com/smartcontractkit/chainlink/v2/core/utils/matches" ) @@ -343,7 +344,7 @@ func TestEngine_AdmissionDenial_StaysSilent(t *testing.T) { //nolint:paralleltes billingClient := metmocks.NewBillingClient(t) // no billing calls expected: admission runs before metering harness := newDropPathHarness(t, billingClient, func(cfg *v2.EngineConfig) { - cfg.Hooks.OnTriggerAdmission = func(context.Context, v2.RoutedTriggerEvent) error { + cfg.Hooks.OnTriggerAdmission = func(context.Context, triggers.CoordinatedEvent) error { return errors.New("not owned by this node") } }) diff --git a/core/services/workflows/v2/engine_test.go b/core/services/workflows/v2/engine_test.go index 0eb775c70a6..a3b73b6e3b3 100644 --- a/core/services/workflows/v2/engine_test.go +++ b/core/services/workflows/v2/engine_test.go @@ -63,6 +63,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncerlimiter" "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" "github.com/smartcontractkit/chainlink/v2/core/utils/matches" ) @@ -917,7 +918,7 @@ func TestEngine_Execution(t *testing.T) { require.NoError(t, <-initDoneCh) // successful trigger registration require.Equal(t, []string{"id_0"}, <-subscribedToTriggersCh) - require.Equal(t, v2.TriggerRegistrationID(cfg.WorkflowID, 0), capturedTriggerRequest.TriggerID) + require.Equal(t, triggers.RegistrationID(cfg.WorkflowID, 0), capturedTriggerRequest.TriggerID) require.Equal(t, cfg.WorkflowID, capturedTriggerRequest.Metadata.WorkflowID) require.Equal(t, cfg.WorkflowOwner, capturedTriggerRequest.Metadata.WorkflowOwner) require.Equal(t, cfg.WorkflowName.Hex(), capturedTriggerRequest.Metadata.WorkflowName) @@ -2501,8 +2502,8 @@ func TestEngine_ExecuteTrigger(t *testing.T) { Workflow: baseCfg.WorkflowID, }) - makeEvent := func(eventID string) v2.RoutedTriggerEvent { - return v2.RoutedTriggerEvent{ + makeEvent := func(eventID string) triggers.CoordinatedEvent { + return triggers.CoordinatedEvent{ WorkflowID: baseCfg.WorkflowID, TriggerCapID: "id_0", TriggerIndex: 0, @@ -2661,13 +2662,13 @@ func TestEngine_ShardDenial(t *testing.T) { module.EXPECT().Close() module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Return(newTriggerSubs(1), nil).Once() }, func(cfg *v2.EngineConfig) { - cfg.Hooks.OnTriggerAdmission = func(_ context.Context, _ v2.RoutedTriggerEvent) error { + cfg.Hooks.OnTriggerAdmission = func(_ context.Context, _ triggers.CoordinatedEvent) error { admissionCalls.Add(1) return tc.wantErr } }) - registrationID := v2.TriggerRegistrationID(baseCfg.WorkflowID, 0) + registrationID := triggers.RegistrationID(baseCfg.WorkflowID, 0) ackedCh := make(chan struct{}, 1) event := capabilities.TriggerEvent{ TriggerType: "basic-trigger@1.0.0", @@ -2893,7 +2894,7 @@ func newTestEngine( OnNodeSynced: func(_ capabilities.Node, _ error) { e.nodeSyncedCalls.Add(1) }, - OnTriggerAdmission: func(_ context.Context, _ v2.RoutedTriggerEvent) error { + OnTriggerAdmission: func(_ context.Context, _ triggers.CoordinatedEvent) error { e.triggerAdmissionCalls.Add(1) return nil }, diff --git a/core/services/workflows/v2/triggers/triggers.go b/core/services/workflows/v2/triggers/triggers.go new file mode 100644 index 00000000000..e3d4012f1d0 --- /dev/null +++ b/core/services/workflows/v2/triggers/triggers.go @@ -0,0 +1,366 @@ +package triggers + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/jonboulle/clockwork" + "golang.org/x/sync/errgroup" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/types/known/anypb" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry" + "github.com/smartcontractkit/chainlink-common/pkg/contexts" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/settings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/validation" + "github.com/smartcontractkit/chainlink/v2/core/platform" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/monitoring" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" +) + +const _prefix = "trigger_reg" + +// Handle is a registered trigger capability plus the registration +// payload/method needed to unregister and re-deliver to it. +type Handle struct { + capabilities.TriggerCapability + Payload *anypb.Any + Method string +} + +// Ack acknowledges a trigger event against the given handle, logging and +// bumping the same success/failure metrics regardless of caller. +// handle may be nil (i.e., registration not found). +func Ack( + ctx context.Context, + lggr logger.Logger, + metrics *monitoring.WorkflowsMetricLabeler, + triggerCapID, triggerRegistrationID, eventID string, + handle *Handle, +) error { + lggr.Infow("ACKing trigger event", "triggerRegistrationID", triggerRegistrationID, "eventID", eventID) + + tm := metrics.With(platform.KeyTriggerID, triggerCapID) + + if handle == nil { + tm.IncrementTriggerEventAckFailureCounter(ctx) + return fmt.Errorf("failed to find trigger %s", triggerRegistrationID) + } + if err := handle.AckEvent(ctx, triggerRegistrationID, eventID, handle.Method); err != nil { + tm.IncrementTriggerEventAckFailureCounter(ctx) + return err + } + tm.IncrementTriggerEventAckSuccessCounter(ctx) + return nil +} + +// CoordinatedEvent is the canonical trigger event type that flows +// through the coordinator path into the engine. +type CoordinatedEvent struct { + WorkflowID string + TriggerCapID string + TriggerIndex int + + // ObservedAt is the time the CoordinatedEvent was constructed. + // It is used for skew metrics (queue wait time) + // and deadline enforcement. + ObservedAt time.Time + + // Deadline is the expiry of this event, + // stamped once at dispatch as ObservedAt + TriggerEventQueueTimeout. + // A settings change after dispatch does not affect already-queued events + Deadline time.Time + + // SequenceNumber determines the execution order of trigger events across the DON. + SequenceNumber uint64 + Event capabilities.TriggerResponse +} + +// ReadLoop consumes triggerEventCh until it closes or ctx is done, +// converting each received capabilities.TriggerResponse into a +// CoordinatedEvent and handing it to deliver. +// +// The caller owns spawning the goroutine, since goroutine lifecycle +// differs between callers. +func ReadLoop( + ctx context.Context, + lggr logger.Logger, + metrics *monitoring.WorkflowsMetricLabeler, + clock clockwork.Clock, + workflowID, triggerCapID string, + triggerIndex int, + triggerEventCh <-chan capabilities.TriggerResponse, + deliver func(ctx context.Context, event CoordinatedEvent), +) { + for { + select { + case <-ctx.Done(): + return + case event, isOpen := <-triggerEventCh: + if !isOpen { + return + } + eventID := event.Event.ID + metrics.With(platform.KeyTriggerID, triggerCapID).IncrementTriggerEventReceivedCounter(ctx) + lggr.Debugw("Processing trigger event", "triggerID", triggerCapID, "eventID", eventID) + if event.Err != nil { + lggr.Errorw("Received a trigger event with error, dropping", "triggerID", triggerCapID, "err", event.Err) + tm := metrics.With(platform.KeyTriggerID, triggerCapID) + tm.IncrementWorkflowTriggerEventErrorCounter(ctx) + tm.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonTriggerResponseError) + continue + } + + routed := CoordinatedEvent{ + WorkflowID: workflowID, + TriggerCapID: triggerCapID, + TriggerIndex: triggerIndex, + ObservedAt: clock.Now(), + Event: event, + } + + deliver(ctx, routed) + } + } +} + +// RegisterDeps bundles what Register needs beyond per-call metadata. +type RegisterDeps struct { + CapRegistry registry.CapabilitiesRegistry + RegTimeout limits.TimeLimiter + ChainAllowed limits.GateLimiter + Settings settings.Getter + Logger logger.Logger + Metrics *monitoring.WorkflowsMetricLabeler +} + +// RegisterMetadata is everything that goes into every trigger's +// capabilities.RequestMetadata for one workflow. +type RegisterMetadata struct { + WorkflowID string // hex-encoded workflow ID + WorkflowOwner string + // WorkflowName is nil-safe: if unset, both WorkflowName and + // DecodedWorkflowName are sent empty rather than panicking on a nil + // interface call. + WorkflowName types.WorkflowName + WorkflowTag string + WorkflowDonID uint32 + // WorkflowDonConfigVersion is the caller's pinned DON config version + // (e.g. engine.go's pinnedWorkflowDonConfigVersion constant). triggers + // does not own this value since it's also used for unrelated things in + // v2 (buildLabels, localNodeSync, capability_executor.go). + WorkflowDonConfigVersion uint32 + WorkflowRegistryChainSelector string + WorkflowRegistryAddress string + // OrgID is the resolved org ID. + OrgID string +} + +// Register concurrently registers subs against their trigger capabilities +// (resolved via deps.CapRegistry), after validating each subscription's +// chain selector and checking chain access for it via deps.ChainAllowed. +// +// On success it returns the registered trigger capability IDs (in +// subscription order), one Handle per registration keyed by registration ID, +// and each registration's event channel. +// +// On any registration failure it unregisters every handle that did succeed +// and returns the first error. +func Register( + ctx context.Context, + deps RegisterDeps, + meta RegisterMetadata, + subs []*sdkpb.TriggerSubscription, +) ([]string, map[string]*Handle, []<-chan capabilities.TriggerResponse, error) { + // check if all requested triggers exist in the registry + tcs := make([]capabilities.TriggerCapability, 0, len(subs)) + for _, sub := range subs { + _, labels, _ := capabilities.ParseID(sub.Id) + chainSelector, err := capabilities.ChainSelectorLabel(labels) + if err != nil { + return nil, nil, nil, fmt.Errorf("invalid chain selector for ID %s: %w", sub.Id, err) + } + if chainSelector != nil { + if allowErr := deps.ChainAllowed.AllowErr(contexts.WithChainSelector(ctx, *chainSelector)); allowErr != nil { + if errors.Is(allowErr, limits.ErrorNotAllowed{}) { + return nil, nil, nil, fmt.Errorf("unable to subscribe to capability %s: ChainSelector %d: %w", sub.Id, *chainSelector, allowErr) + } + return nil, nil, nil, fmt.Errorf("failed to check access for ChainSelector %d: %w", *chainSelector, allowErr) + } + } + triggerCap, err := deps.CapRegistry.GetTrigger(ctx, sub.Id) + if err != nil { + return nil, nil, nil, fmt.Errorf("trigger capability not found: %w", err) + } + tcs = append(tcs, triggerCap) + } + + // register to all triggers concurrently + regCtx, regCancel, err := deps.RegTimeout.WithTimeout(ctx) + if err != nil { + return nil, nil, nil, err + } + defer regCancel() + + // trigger registration results for use in concurrent trigger subscriptions + type triggerRegResult struct { + index int + registrationID string + triggerCap capabilities.TriggerCapability + eventCh <-chan capabilities.TriggerResponse + payload *anypb.Any + method string + triggerCapID string + } + + resultsCh := make(chan triggerRegResult, len(subs)) + g, gCtx := errgroup.WithContext(regCtx) + + var workflowNameHex, decodedWorkflowName string + if meta.WorkflowName != nil { + workflowNameHex = meta.WorkflowName.Hex() + decodedWorkflowName = meta.WorkflowName.String() + } + + // Launch concurrent trigger registrations + for i, sub := range subs { + triggerCap := tcs[i] + g.Go(func() error { + registrationID := RegistrationID(meta.WorkflowID, i) + args := []any{"triggerID", sub.Id, "method", sub.Method} + if sub.Payload != nil { + args = append(args, "payload", protojson.Format(sub.Payload)) + } + deps.Logger.Infow("Registering trigger", args...) + metadata := capabilities.RequestMetadata{ + WorkflowID: meta.WorkflowID, + WorkflowOwner: meta.WorkflowOwner, + WorkflowName: workflowNameHex, + DecodedWorkflowName: decodedWorkflowName, + WorkflowTag: meta.WorkflowTag, + WorkflowDonID: meta.WorkflowDonID, + WorkflowDonConfigVersion: meta.WorkflowDonConfigVersion, + ReferenceID: fmt.Sprintf("trigger_%d", i), + WorkflowRegistryChainSelector: meta.WorkflowRegistryChainSelector, + WorkflowRegistryAddress: meta.WorkflowRegistryAddress, + EngineVersion: platform.ValueWorkflowVersionV2, + // no WorkflowExecutionID needed (or available at this stage) + } + propagateOrgIDMeta, _ := cresettings.Default.PropagateOrgIDInRequestMetadata.GetOrDefault(gCtx, deps.Settings) + if propagateOrgIDMeta && meta.OrgID != "" { + metadata.OrgID = meta.OrgID + } + triggerEventCh, regErr := triggerCap.RegisterTrigger(gCtx, capabilities.TriggerRegistrationRequest{ + TriggerID: registrationID, + Metadata: metadata, + Payload: sub.Payload, + Method: sub.Method, + // no Config needed - NoDAG uses Payload + }) + if regErr != nil { + deps.Logger.Errorw("Trigger registration failed", "triggerID", sub.Id, "err", regErr) + deps.Metrics.With(platform.KeyTriggerID, sub.Id).IncrementRegisterTriggerFailureCounter(gCtx) + return fmt.Errorf("failed to register trigger %s: %w", sub.Id, regErr) + } + // Send successful result + resultsCh <- triggerRegResult{ + index: i, + registrationID: registrationID, + triggerCap: triggerCap, + eventCh: triggerEventCh, + payload: sub.Payload, + method: sub.Method, + triggerCapID: sub.Id, + } + return nil + }) + } + + // wait for all registrations to complete. + // returns first non-nil error. + registrationErr := g.Wait() + close(resultsCh) + + handles := make(map[string]*Handle, len(subs)) + eventChans := make([]<-chan capabilities.TriggerResponse, len(subs)) + triggerCapIDs := make([]string, len(subs)) + for result := range resultsCh { + handles[result.registrationID] = &Handle{ + TriggerCapability: result.triggerCap, + Payload: result.payload, + Method: result.method, + } + eventChans[result.index] = result.eventCh + triggerCapIDs[result.index] = result.triggerCapID + } + + // If any registration failed, unregister successful ones and return error + if registrationErr != nil { + deps.Logger.Errorw("One or more trigger registrations failed - reverting all", "err", registrationErr) + Unregister(ctx, deps.Logger, meta.WorkflowID, meta.WorkflowDonID, handles) + return nil, nil, nil, registrationErr + } + + deps.Logger.Infow("All triggers registered successfully", "numTriggers", len(subs), "triggerIDs", triggerCapIDs) + deps.Metrics.IncrementWorkflowRegisteredCounter(ctx) + return triggerCapIDs, handles, eventChans, nil +} + +// Unregister unregisters every handle in handles with the capability +// registry. It logs individual UnregisterTrigger failures and returns how +// many occurred, so callers can log their own summary. +func Unregister(ctx context.Context, lggr logger.Logger, workflowID string, workflowDonID uint32, handles map[string]*Handle) (failCount int) { + for registrationID, handle := range handles { + if err := handle.UnregisterTrigger(ctx, capabilities.TriggerRegistrationRequest{ + TriggerID: registrationID, + Metadata: capabilities.RequestMetadata{ + WorkflowID: workflowID, + WorkflowDonID: workflowDonID, + }, + Payload: handle.Payload, + Method: handle.Method, + }); err != nil { + lggr.Errorw("Failed to unregister trigger", "registrationId", registrationID, "err", err) + failCount++ + } + } + return failCount +} + +// RegistrationID constructs a trigger registration ID from a workflow ID and trigger index. +func RegistrationID(workflowID string, triggerIndex int) string { + return fmt.Sprintf("%s_%s_%d", _prefix, workflowID, triggerIndex) +} + +// ParseWorkflowID extracts the workflow ID from a registration ID produced by RegistrationID. +func ParseWorkflowID(registrationID string) (string, error) { + rest, ok := strings.CutPrefix(registrationID, _prefix+"_") + if !ok { + return "", fmt.Errorf("invalid trigger registration ID %q: missing prefix %q", registrationID, _prefix) + } + + idx := strings.LastIndex(rest, "_") + if idx == -1 { + return "", fmt.Errorf("invalid trigger registration ID %q: missing trigger index", registrationID) + } + + workflowID, triggerIndexStr := rest[:idx], rest[idx+1:] + if _, err := strconv.Atoi(triggerIndexStr); err != nil { + return "", fmt.Errorf("invalid trigger registration ID %q: invalid trigger index %q: %w", registrationID, triggerIndexStr, err) + } + + if err := validation.ValidateWorkflowOrExecutionID(workflowID); err != nil { + return "", fmt.Errorf("invalid trigger registration ID %q: invalid workflow ID %q: %w", registrationID, workflowID, err) + } + + return workflowID, nil +} diff --git a/core/services/workflows/v2/triggers/triggers_test.go b/core/services/workflows/v2/triggers/triggers_test.go new file mode 100644 index 00000000000..85a504fb0c4 --- /dev/null +++ b/core/services/workflows/v2/triggers/triggers_test.go @@ -0,0 +1,520 @@ +package triggers + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "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/metrics" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" + regmocks "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks" + sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + capmocks "github.com/smartcontractkit/chainlink/v2/core/capabilities/mocks" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/monitoring" +) + +// validWorkflowID is a syntactically valid workflow ID: a 64-character hex string (32 bytes). +var validWorkflowID = strings.Repeat("ab", 32) + +func TestRegistrationID(t *testing.T) { + t.Parallel() + tests := []struct { + name string + workflowID string + triggerIndex int + want string + }{ + { + name: "simple workflow ID", + workflowID: "abc123", + triggerIndex: 0, + want: "trigger_reg_abc123_0", + }, + { + name: "workflow ID containing underscores", + workflowID: "wf_with_underscores", + triggerIndex: 5, + want: "trigger_reg_wf_with_underscores_5", + }, + { + name: "empty workflow ID", + workflowID: "", + triggerIndex: 2, + want: "trigger_reg__2", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := RegistrationID(tt.workflowID, tt.triggerIndex) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestParseWorkflowID(t *testing.T) { + t.Parallel() + tests := []struct { + name string + registrationID string + wantWorkflowID string + wantErrSubstring string + }{ + { + name: "valid workflow ID", + registrationID: "trigger_reg_" + validWorkflowID + "_0", + wantWorkflowID: validWorkflowID, + }, + { + name: "valid workflow ID with large trigger index", + registrationID: "trigger_reg_" + validWorkflowID + "_12345", + wantWorkflowID: validWorkflowID, + }, + { + name: "missing prefix", + registrationID: "not_a_registration_id_0", + wantErrSubstring: "missing prefix", + }, + { + name: "missing trigger index", + registrationID: "trigger_reg_" + validWorkflowID, + wantErrSubstring: "missing trigger index", + }, + { + name: "non-numeric trigger index", + registrationID: "trigger_reg_" + validWorkflowID + "_notanumber", + wantErrSubstring: "invalid trigger index", + }, + { + name: "empty workflow ID", + registrationID: "trigger_reg__2", + wantErrSubstring: "invalid workflow ID", + }, + { + name: "workflow ID too short", + registrationID: "trigger_reg_abc123_0", + wantErrSubstring: "invalid workflow ID", + }, + { + name: "workflow ID not hex-encoded", + registrationID: "trigger_reg_" + strings.Repeat("z", 64) + "_0", + wantErrSubstring: "invalid workflow ID", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := ParseWorkflowID(tt.registrationID) + if tt.wantErrSubstring != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErrSubstring) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantWorkflowID, got) + }) + } +} + +func newTestMetrics(t *testing.T) *monitoring.WorkflowsMetricLabeler { + t.Helper() + em, err := monitoring.InitMonitoringResources() + require.NoError(t, err) + return monitoring.NewWorkflowsMetricLabeler(metrics.NewLabeler(), em) +} + +func TestAck(t *testing.T) { + t.Parallel() + const ( + triggerCapID = "trigger-cap-id" + triggerRegistrationID = "trigger-reg-id" + eventID = "event-id" + method = "test-method" + ) + + t.Run("returns an error when the handle is not found", func(t *testing.T) { + t.Parallel() + err := Ack(t.Context(), logger.Test(t), newTestMetrics(t), triggerCapID, triggerRegistrationID, eventID, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), triggerRegistrationID) + }) + + t.Run("delegates to the handle's AckEvent using its stored method", func(t *testing.T) { + t.Parallel() + trigger := capmocks.NewTriggerCapability(t) + trigger.EXPECT().AckEvent(mock.Anything, triggerRegistrationID, eventID, method).Return(nil).Once() + handle := &Handle{TriggerCapability: trigger, Method: method} + + err := Ack(t.Context(), logger.Test(t), newTestMetrics(t), triggerCapID, triggerRegistrationID, eventID, handle) + require.NoError(t, err) + }) + + t.Run("propagates the handle's AckEvent error", func(t *testing.T) { + t.Parallel() + wantErr := errors.New("boom") + trigger := capmocks.NewTriggerCapability(t) + trigger.EXPECT().AckEvent(mock.Anything, triggerRegistrationID, eventID, method).Return(wantErr).Once() + handle := &Handle{TriggerCapability: trigger, Method: method} + + err := Ack(t.Context(), logger.Test(t), newTestMetrics(t), triggerCapID, triggerRegistrationID, eventID, handle) + require.ErrorIs(t, err, wantErr) + }) +} + +func TestReadLoop(t *testing.T) { + t.Parallel() + const ( + workflowID = "wf-id" + triggerCapID = "trigger-cap-id" + triggerIndex = 3 + chanBuf = 4 + ) + + tests := []struct { + name string + // wantDelivered are the event IDs ReadLoop is expected to hand to + // deliver, in order. Nil means deliver must not be called at all. + wantDelivered []string + // run drives the scenario: it sends/closes on triggerEventCh and/or + // cancels the reader's context. + run func(triggerEventCh chan<- capabilities.TriggerResponse, cancel context.CancelFunc) + }{ + { + name: "delivers a routed event built from the response and the clock", + wantDelivered: []string{"event-1"}, + run: func(triggerEventCh chan<- capabilities.TriggerResponse, _ context.CancelFunc) { + triggerEventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{ID: "event-1"}} + }, + }, + { + name: "skips delivery when the response carries an error, but keeps consuming", + wantDelivered: []string{"event-2"}, + run: func(triggerEventCh chan<- capabilities.TriggerResponse, _ context.CancelFunc) { + triggerEventCh <- capabilities.TriggerResponse{Err: errors.New("boom")} + triggerEventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{ID: "event-2"}} + }, + }, + { + name: "delivers multiple events in order", + wantDelivered: []string{"a", "b"}, + run: func(triggerEventCh chan<- capabilities.TriggerResponse, _ context.CancelFunc) { + triggerEventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{ID: "a"}} + triggerEventCh <- capabilities.TriggerResponse{Event: capabilities.TriggerEvent{ID: "b"}} + }, + }, + { + name: "returns when the trigger channel closes", + run: func(triggerEventCh chan<- capabilities.TriggerResponse, _ context.CancelFunc) { + close(triggerEventCh) + }, + }, + { + name: "returns when ctx is canceled", + run: func(_ chan<- capabilities.TriggerResponse, cancel context.CancelFunc) { + cancel() + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + // A single deadline governs the whole subtest: every wait below + // (deliveries, then ReadLoop's return) races against this one + // timer instead of each getting its own fresh timeout. + deadline, cancelDeadline := context.WithTimeout(context.Background(), 5*time.Second) + defer cancelDeadline() + + sutCtx, sutCancel := context.WithCancel(deadline) + defer sutCancel() + + triggerEventCh := make(chan capabilities.TriggerResponse, chanBuf) + deliverCh := make(chan CoordinatedEvent, chanBuf) + deliver := func(_ context.Context, event CoordinatedEvent) { + deliverCh <- event + } + + clock := clockwork.NewFakeClock() + lggr := logger.Test(t) + testMetrics := newTestMetrics(t) + done := make(chan struct{}) + go func() { + defer close(done) + ReadLoop(sutCtx, lggr, testMetrics, clock, workflowID, triggerCapID, triggerIndex, triggerEventCh, deliver) + }() + + tt.run(triggerEventCh, sutCancel) + + var gotDelivered []string + for range tt.wantDelivered { + select { + case event := <-deliverCh: + assert.Equal(t, workflowID, event.WorkflowID) + assert.Equal(t, triggerCapID, event.TriggerCapID) + assert.Equal(t, triggerIndex, event.TriggerIndex) + assert.Equal(t, clock.Now(), event.ObservedAt) + gotDelivered = append(gotDelivered, event.Event.Event.ID) + case <-deadline.Done(): + t.Fatal("timed out waiting for delivery") + } + } + assert.Equal(t, tt.wantDelivered, gotDelivered) + + // Whatever the scenario did to end ReadLoop (close/cancel), make + // sure it actually returns before the shared deadline. + sutCancel() + select { + case <-done: + case <-deadline.Done(): + t.Fatal("ReadLoop did not return before the test deadline") + } + }) + } +} + +func newRegisterDeps(t *testing.T, capReg registry.CapabilitiesRegistry, chainAllowed limits.GateLimiter) RegisterDeps { + t.Helper() + return RegisterDeps{ + CapRegistry: capReg, + RegTimeout: limits.NewTimeLimiter(5 * time.Second), + ChainAllowed: chainAllowed, + Logger: logger.Test(t), + Metrics: newTestMetrics(t), + } +} + +// erroringTimeLimiter is a limits.TimeLimiter whose WithTimeout always fails, +// for exercising Register's early-return path when the registration timeout +// itself can't be established. +type erroringTimeLimiter struct{} + +func (erroringTimeLimiter) Close() error { return nil } +func (erroringTimeLimiter) Limit(context.Context) (time.Duration, error) { + return 0, nil +} +func (erroringTimeLimiter) WithTimeout(context.Context) (context.Context, func(), error) { + return nil, nil, errors.New("timeout limiter broken") +} + +func TestRegister(t *testing.T) { + t.Parallel() + const ( + workflowID = "wf-id" + workflowOwner = "wf-owner" + workflowDonID = uint32(7) + donConfigVer = uint32(3) + ) + + baseMeta := RegisterMetadata{ + WorkflowID: workflowID, + WorkflowOwner: workflowOwner, + WorkflowDonID: workflowDonID, + WorkflowDonConfigVersion: donConfigVer, + } + + t.Run("registers all subscriptions and returns handles, capability IDs, and event channels in subscription order", func(t *testing.T) { + t.Parallel() + sub0 := &sdkpb.TriggerSubscription{Id: "trigger-a@1.0.0", Method: "method-a"} + sub1 := &sdkpb.TriggerSubscription{Id: "trigger-b@1.0.0", Method: "method-b"} + + ch0 := make(chan capabilities.TriggerResponse) + ch1 := make(chan capabilities.TriggerResponse) + + var mu sync.Mutex + gotMetadata := map[string]capabilities.RequestMetadata{} + captureMetadata := func(triggerCapID string) func(context.Context, capabilities.TriggerRegistrationRequest) { + return func(_ context.Context, req capabilities.TriggerRegistrationRequest) { + mu.Lock() + gotMetadata[triggerCapID] = req.Metadata + mu.Unlock() + } + } + + trigger0 := capmocks.NewTriggerCapability(t) + trigger0.EXPECT().RegisterTrigger(mock.Anything, mock.Anything). + Run(captureMetadata(sub0.Id)). + Return((<-chan capabilities.TriggerResponse)(ch0), nil).Once() + trigger1 := capmocks.NewTriggerCapability(t) + trigger1.EXPECT().RegisterTrigger(mock.Anything, mock.Anything). + Run(captureMetadata(sub1.Id)). + Return((<-chan capabilities.TriggerResponse)(ch1), nil).Once() + + capReg := regmocks.NewCapabilitiesRegistry(t) + capReg.EXPECT().GetTrigger(mock.Anything, sub0.Id).Return(trigger0, nil).Once() + capReg.EXPECT().GetTrigger(mock.Anything, sub1.Id).Return(trigger1, nil).Once() + + deps := newRegisterDeps(t, capReg, limits.NewGateLimiter(true)) + triggerCapIDs, handles, eventChans, err := Register(t.Context(), deps, baseMeta, []*sdkpb.TriggerSubscription{sub0, sub1}) + require.NoError(t, err) + + assert.Equal(t, []string{sub0.Id, sub1.Id}, triggerCapIDs) + require.Len(t, eventChans, 2) + assert.Equal(t, (<-chan capabilities.TriggerResponse)(ch0), eventChans[0]) + assert.Equal(t, (<-chan capabilities.TriggerResponse)(ch1), eventChans[1]) + + reg0, reg1 := RegistrationID(workflowID, 0), RegistrationID(workflowID, 1) + require.Contains(t, handles, reg0) + require.Contains(t, handles, reg1) + assert.Equal(t, sub0.Method, handles[reg0].Method) + assert.Equal(t, sub1.Method, handles[reg1].Method) + + mu.Lock() + defer mu.Unlock() + md0 := gotMetadata[sub0.Id] + assert.Equal(t, workflowID, md0.WorkflowID) + assert.Equal(t, workflowOwner, md0.WorkflowOwner) + assert.Equal(t, workflowDonID, md0.WorkflowDonID) + assert.Equal(t, donConfigVer, md0.WorkflowDonConfigVersion) + assert.Equal(t, "trigger_0", md0.ReferenceID) + assert.Empty(t, md0.OrgID, "OrgID propagation is off by default") + }) + + t.Run("returns an error without registering when the chain gate is closed", func(t *testing.T) { + t.Parallel() + sub := &sdkpb.TriggerSubscription{Id: "trigger-a:ChainSelector_12345@1.0.0", Method: "method-a"} + // No GetTrigger expectation: the gate must deny before the registry is consulted. + capReg := regmocks.NewCapabilitiesRegistry(t) + deps := newRegisterDeps(t, capReg, limits.NewGateLimiter(false)) + + _, _, _, err := Register(t.Context(), deps, baseMeta, []*sdkpb.TriggerSubscription{sub}) + require.Error(t, err) + assert.Contains(t, err.Error(), "ChainSelector") + }) + + t.Run("returns an error when a trigger capability isn't found in the registry", func(t *testing.T) { + t.Parallel() + sub := &sdkpb.TriggerSubscription{Id: "trigger-a@1.0.0", Method: "method-a"} + capReg := regmocks.NewCapabilitiesRegistry(t) + capReg.EXPECT().GetTrigger(mock.Anything, sub.Id).Return(nil, errors.New("not found")).Once() + deps := newRegisterDeps(t, capReg, limits.NewGateLimiter(true)) + + _, _, _, err := Register(t.Context(), deps, baseMeta, []*sdkpb.TriggerSubscription{sub}) + require.Error(t, err) + assert.Contains(t, err.Error(), "trigger capability not found") + }) + + t.Run("unregisters already-succeeded handles when one registration fails", func(t *testing.T) { + t.Parallel() + sub0 := &sdkpb.TriggerSubscription{Id: "trigger-a@1.0.0", Method: "method-a"} + sub1 := &sdkpb.TriggerSubscription{Id: "trigger-b@1.0.0", Method: "method-b"} + + ch0 := make(chan capabilities.TriggerResponse) + trigger0 := capmocks.NewTriggerCapability(t) + trigger0.EXPECT().RegisterTrigger(mock.Anything, mock.Anything). + Return((<-chan capabilities.TriggerResponse)(ch0), nil).Once() + trigger0.EXPECT().UnregisterTrigger(mock.Anything, mock.Anything).Return(nil).Once() + + trigger1 := capmocks.NewTriggerCapability(t) + trigger1.EXPECT().RegisterTrigger(mock.Anything, mock.Anything). + Return(nil, errors.New("registration failed")).Once() + + capReg := regmocks.NewCapabilitiesRegistry(t) + capReg.EXPECT().GetTrigger(mock.Anything, sub0.Id).Return(trigger0, nil).Once() + capReg.EXPECT().GetTrigger(mock.Anything, sub1.Id).Return(trigger1, nil).Once() + + deps := newRegisterDeps(t, capReg, limits.NewGateLimiter(true)) + triggerCapIDs, handles, eventChans, err := Register(t.Context(), deps, baseMeta, []*sdkpb.TriggerSubscription{sub0, sub1}) + + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to register trigger") + assert.Nil(t, triggerCapIDs) + assert.Nil(t, handles) + assert.Nil(t, eventChans) + // trigger0.UnregisterTrigger's expectation being satisfied (verified by + // capmocks.NewTriggerCapability's t.Cleanup) confirms the rollback ran. + }) + + t.Run("returns an error when the registration timeout limiter fails", func(t *testing.T) { + t.Parallel() + sub := &sdkpb.TriggerSubscription{Id: "trigger-a@1.0.0", Method: "method-a"} + trigger0 := capmocks.NewTriggerCapability(t) + capReg := regmocks.NewCapabilitiesRegistry(t) + capReg.EXPECT().GetTrigger(mock.Anything, sub.Id).Return(trigger0, nil).Once() + + deps := newRegisterDeps(t, capReg, limits.NewGateLimiter(true)) + deps.RegTimeout = erroringTimeLimiter{} + + _, _, _, err := Register(t.Context(), deps, baseMeta, []*sdkpb.TriggerSubscription{sub}) + require.Error(t, err) + assert.Contains(t, err.Error(), "timeout limiter broken") + }) +} + +func TestUnregister(t *testing.T) { + t.Parallel() + const ( + workflowID = "wf-id" + workflowDonID = uint32(9) + ) + + t.Run("unregisters every handle and returns zero failures on success", func(t *testing.T) { + t.Parallel() + trigger0 := capmocks.NewTriggerCapability(t) + trigger0.EXPECT().UnregisterTrigger(mock.Anything, mock.MatchedBy(func(req capabilities.TriggerRegistrationRequest) bool { + return req.TriggerID == "reg-0" && req.Metadata.WorkflowID == workflowID && req.Metadata.WorkflowDonID == workflowDonID + })).Return(nil).Once() + trigger1 := capmocks.NewTriggerCapability(t) + trigger1.EXPECT().UnregisterTrigger(mock.Anything, mock.Anything).Return(nil).Once() + + handles := map[string]*Handle{ + "reg-0": {TriggerCapability: trigger0, Method: "m0"}, + "reg-1": {TriggerCapability: trigger1, Method: "m1"}, + } + + failCount := Unregister(t.Context(), logger.Test(t), workflowID, workflowDonID, handles) + assert.Equal(t, 0, failCount) + }) + + t.Run("counts and logs individual unregister failures", func(t *testing.T) { + t.Parallel() + trigger0 := capmocks.NewTriggerCapability(t) + trigger0.EXPECT().UnregisterTrigger(mock.Anything, mock.Anything).Return(errors.New("boom")).Once() + + handles := map[string]*Handle{"reg-0": {TriggerCapability: trigger0, Method: "m0"}} + + failCount := Unregister(t.Context(), logger.Test(t), workflowID, workflowDonID, handles) + assert.Equal(t, 1, failCount) + }) + + t.Run("returns zero for an empty handle map", func(t *testing.T) { + t.Parallel() + failCount := Unregister(t.Context(), logger.Test(t), workflowID, workflowDonID, map[string]*Handle{}) + assert.Equal(t, 0, failCount) + }) +} + +func TestRegistrationID_ParseWorkflowID_RoundTrip(t *testing.T) { + t.Parallel() + tests := []struct { + name string + workflowID string + triggerIndex int + }{ + {name: "trigger index zero", workflowID: validWorkflowID, triggerIndex: 0}, + {name: "large trigger index", workflowID: validWorkflowID, triggerIndex: 12345}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + registrationID := RegistrationID(tt.workflowID, tt.triggerIndex) + + got, err := ParseWorkflowID(registrationID) + require.NoError(t, err) + assert.Equal(t, tt.workflowID, got) + }) + } +} diff --git a/core/services/workflows/v2/types.go b/core/services/workflows/v2/types.go index 83339190735..bcc48db169a 100644 --- a/core/services/workflows/v2/types.go +++ b/core/services/workflows/v2/types.go @@ -4,10 +4,10 @@ import ( "context" "time" - "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/contexts" "github.com/smartcontractkit/chainlink-common/pkg/services" sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2/triggers" ) // EventSink is how trigger events are delivered to an engine for execution. @@ -33,7 +33,7 @@ import ( // NOT returned as errors. They are captured by OnExecutionError and // OnExecutionFinished hooks. ExecuteTrigger returns nil in these cases. type EventSink interface { - ExecuteTrigger(ctx context.Context, event RoutedTriggerEvent) error + ExecuteTrigger(ctx context.Context, event triggers.CoordinatedEvent) error } // Acknowledger acknowledges a trigger event without the engine owning the @@ -87,25 +87,3 @@ type WorkflowEngine interface { // Fixed at construction. IsCoordinated() bool } - -// RoutedTriggerEvent is the canonical trigger event type that flows -// through the coordinator path into the engine. -type RoutedTriggerEvent struct { - WorkflowID string - TriggerCapID string - TriggerIndex int - - // ObservedAt is the time the RoutedTriggerEvent was constructed by the - // coordinator. It is used for skew metrics (queue wait time) - // and deadline enforcement. - ObservedAt time.Time - - // Deadline is the expiry of this event in the coordinator queue, - // stamped once at dispatch as ObservedAt + TriggerEventQueueTimeout. - // A settings change after dispatch does not affect already-queued events - Deadline time.Time - - // SequenceNumber determines the execution order of trigger events across the DON. In M1 it is always 0 (no consensus ordering). - SequenceNumber uint64 - Event capabilities.TriggerResponse -}