[pull] develop from smartcontractkit:develop

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

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)
-}

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) {}

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)
 }
 

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,