Add direct module cache priming for stored trigger subs

core/services/workflows/v2/base_engine.go

@@ -32,6 +32,7 @@
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows"
+	workflowhost "github.com/smartcontractkit/chainlink-common/pkg/workflows/host"
 	billing "github.com/smartcontractkit/chainlink-protos/billing/go"
 	sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
 	protoevents "github.com/smartcontractkit/chainlink-protos/workflows/go/events"
@@ -226,6 +227,13 @@
 		if err := e.cfg.LocalLimiters.TriggerSubscription.Check(ctx, len(e.cfg.CachedTriggerSubscriptions)); err != nil {
 			return nil, err
 		}
+		// The cache path skips Execute(Subscribe), so modules that derive their trigger
+		// routing from that call (e.g. requirementSelectingModule) must be primed explicitly.
+		if primer, ok := e.cfg.Module.(workflowhost.TriggerCachePrimer); ok {
+			if err := primer.PrimeTriggerCache(ctx, e.cfg.CachedTriggerSubscriptions); err != nil {
+				return nil, fmt.Errorf("failed to prime trigger cache: %w", err)
+			}
+		}
 		e.metrics.IncrementTriggerSubscriptionSourceCounter(ctx, triggerSubscriptionSourceCache)
 		if err := e.cfg.Hooks.OnSubscriptionsReady(e.cfg.CachedTriggerSubscriptions, e.Tenant(), true); err != nil {
 			return nil, fmt.Errorf("OnSubscriptionsReady hook failed: %w", err)

core/services/workflows/v2/engine_test.go

@@ -46,6 +46,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	regmocks "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows"
+	generichost "github.com/smartcontractkit/chainlink-common/pkg/workflows/host"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host"
 	modulemocks "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/mocks"
 	billing "github.com/smartcontractkit/chainlink-protos/billing/go"
@@ -700,6 +701,88 @@ func newTriggerSubs(n int) *sdkpb.ExecutionResult {
 	}
 }
 
+// stubExecuteModule is a minimal generichost.Module used to exercise the real
+// requirementSelectingModule wiring (modulemocks.ModuleV2 doesn't implement
+// generichost.TriggerCachePrimer, so it can't catch a regression here).
+type stubExecuteModule struct {
+	executeFn func(context.Context, *sdkpb.ExecuteRequest, generichost.ExecutionHelper) (*sdkpb.ExecutionResult, error)
+}
+
+func (s *stubExecuteModule) Start() {}
+func (s *stubExecuteModule) Close() {}
+func (s *stubExecuteModule) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, h generichost.ExecutionHelper) (*sdkpb.ExecutionResult, error) {
+	return s.executeFn(ctx, req, h)
+}
+
+// TestEngine_ExecuteTrigger_AfterCachedSubscriptions guards against a regression where
+// enabling CachedTriggerSubscriptionsEnabled made every trigger execution fail with
+// "cannot trigger before gathering subscriptions": the cache path skipped Execute(Subscribe),
+// which left generichost.requirementSelectingModule's trigger-routing cache unprimed.
+func TestEngine_ExecuteTrigger_AfterCachedSubscriptions(t *testing.T) {
+	t.Parallel()
+
+	cachedSubs := []*sdkpb.TriggerSubscription{
+		{Id: "id_0", Method: "method"},
+		{Id: "id_1", Method: "method"},
+	}
+
+	stub := &stubExecuteModule{
+		executeFn: func(_ context.Context, req *sdkpb.ExecuteRequest, _ generichost.ExecutionHelper) (*sdkpb.ExecutionResult, error) {
+			if req.GetTrigger() != nil {
+				return &sdkpb.ExecutionResult{Result: &sdkpb.ExecutionResult_Value{}}, nil
+			}
+			t.Fatal("Subscribe should never execute on the cached-subscriptions path")
+			return nil, nil
+		},
+	}
+	module := generichost.NewRequirementSelectingModule(generichost.ModuleAndHandler{Module: stub}, nil)
+
+	capreg := regmocks.NewCapabilitiesRegistry(t)
+	capreg.EXPECT().LocalNode(matches.AnyContext).Return(newNode(t), nil)
+	trigger0, trigger1 := capmocks.NewTriggerCapability(t), capmocks.NewTriggerCapability(t)
+	capreg.EXPECT().GetTrigger(matches.AnyContext, "id_0").Return(trigger0, nil).Once()
+	capreg.EXPECT().GetTrigger(matches.AnyContext, "id_1").Return(trigger1, nil).Once()
+	tr0Ch, tr1Ch := make(chan capabilities.TriggerResponse), make(chan capabilities.TriggerResponse)
+	trigger0.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(tr0Ch, nil).Once()
+	trigger1.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(tr1Ch, nil).Once()
+	trigger0.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once()
+	trigger1.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once()
+	trigger0.EXPECT().AckEvent(matches.AnyContext, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
+
+	cfg := defaultTestConfig(t, nil)
+	cfg.Module = module
+	cfg.CapRegistry = capreg
+	cfg.BillingClient = setupMockBillingClient(t)
+	cfg.CachedTriggerSubscriptions = cachedSubs
+	cfg.CachedTriggerSubscriptionsEnabled = true
+
+	finishedCh := make(chan string, 1)
+	initDoneCh := make(chan error, 1)
+	cfg.Hooks = v2.LifecycleHooks{
+		OnInitialized:       func(err error) { initDoneCh <- err },
+		OnExecutionFinished: func(_, status string) { finishedCh <- status },
+	}
+
+	engine, err := v2.NewEngine(cfg)
+	require.NoError(t, err)
+	servicetest.Run(t, engine)
+	require.NoError(t, <-initDoneCh)
+
+	ctx := contexts.WithCRE(t.Context(), contexts.CRE{Owner: cfg.WorkflowOwner, Workflow: cfg.WorkflowID})
+	event := triggers.CoordinatedEvent{
+		WorkflowID:   cfg.WorkflowID,
+		TriggerCapID: "id_0",
+		TriggerIndex: 0,
+		ObservedAt:   time.Now(),
+		Event: capabilities.TriggerResponse{
+			Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "cached_trigger_event"},
+		},
+	}
+	err = engine.ExecuteTrigger(ctx, event)
+	require.NoError(t, err)
+	require.Equal(t, "completed", <-finishedCh)
+}
+
 func TestEngine_OrganizationIdLogger(t *testing.T) {
 	t.Parallel()