From dfe1fbc81c0f6c87ec31772eb5b603edbfdda0bb Mon Sep 17 00:00:00 2001 From: Bolek Kulbabinski <1416262+bolekk@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:49:28 -0700 Subject: [PATCH] Add direct module cache priming for stored trigger subs --- core/services/workflows/v2/base_engine.go | 8 +++ core/services/workflows/v2/engine_test.go | 83 +++++++++++++++++++++++ 2 files changed, 91 insertions(+) diff --git a/core/services/workflows/v2/base_engine.go b/core/services/workflows/v2/base_engine.go index 8a09e7f6ca1..01afdc0fe53 100644 --- a/core/services/workflows/v2/base_engine.go +++ b/core/services/workflows/v2/base_engine.go @@ -32,6 +32,7 @@ import ( "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 @@ func (e *baseEngine) Subscribe(ctx context.Context) ([]*sdkpb.TriggerSubscriptio 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) diff --git a/core/services/workflows/v2/engine_test.go b/core/services/workflows/v2/engine_test.go index 90bb251b09e..5d1b59c258c 100644 --- a/core/services/workflows/v2/engine_test.go +++ b/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()