Local CRE: Use Centralized Service CTF Component (Local StorageService)

core/scripts/cre/environment/configs/extensions/README.md

@@ -0,0 +1,22 @@
+# Local CRE extensions
+
+Centralized CRE services run from their own repos' compose files, through their chainlink-testing-framework components
+(`framework/components/dockercompose/cre_storage_service`). With an `[extensions]` section (`extensions.toml`),
+`env start` starts them on `ctf` after the DONs, passing the registry chain, its CapabilitiesRegistry and the workflow
+DON family, and points the workflow nodes at the roles they answer to. `env stop` removes them.
+
+| Role (alias on `ctf`) | Port on `ctf` | Workflow node config |
+| --- | --- | --- |
+| `cre-workflow-source` | 50052 | `Capabilities.WorkflowRegistry.AdditionalSources` |
+| `cre-artifact-api` | 50052 | `Capabilities.WorkflowRegistry.WorkflowStorage.URL` |
+| `cre-artifact-store` | 4566 | `WorkflowStorage.ArtifactStorageHost`, allowed for the gateway |
+| `cre-deploy-api` | 50051 | deploy tooling (`registryclient`) |
+
+`storage_compose` defaults to a `cre-storage-service` checkout next to this repo; its images are built locally (see its
+`local-stack/README.md`). Workflows then come from the extensions, not `env workflow deploy`; deploy tooling runs in a
+container on `ctf`, since artifact URLs use `cre-artifact-store`. From the host, the roles are on free ports that
+`env start` logs.
+
+```bash
+CTF_CONFIGS=configs/workflow-gateway-don.toml,configs/extensions/extensions.toml go run . env start
+```

core/scripts/cre/environment/configs/extensions/extensions.toml

@@ -0,0 +1,3 @@
+# Starts the extension stacks (README.md) with any topology. Add it after the topology in CTF_CONFIGS.
+[extensions]
+  storage_compose = "../../../../../cre-storage-service/local-stack/compose.yaml"

core/scripts/cre/environment/environment/environment.go

@@ -348,6 +348,14 @@ func startCmd() *cobra.Command {
 				return errors.Wrap(err, "failed to validate test configuration")
 			}
 
+			var extensionsDonFamily string
+			if in.Extensions != nil {
+				var extErr error
+				if extensionsDonFamily, extErr = validateExtensions(cmdContext, in); extErr != nil {
+					return errors.Wrap(extErr, "invalid extensions config")
+				}
+			}
+
 			topologySummary, _, topErr := generateTopologyArtifactsForLoadedConfig(in)
 			if topErr != nil {
 				framework.L.Warn().Err(topErr).Msg("failed to generate topology visualization artifacts")
@@ -401,6 +409,12 @@ func startCmd() *cobra.Command {
 				fmt.Fprintf(os.Stderr, "failed to track startup: %s\n", dxErr)
 			}
 
+			if in.Extensions != nil {
+				if err := startExtensions(cmdContext, in.Extensions, extensionsDonFamily, output); err != nil {
+					return errors.Wrap(err, "failed to start extensions")
+				}
+			}
+
 			if withChipIngressStack {
 				startChipIngressStackErr := startChipIngressStack(
 					cmdContext,
@@ -643,6 +657,10 @@ func stopCmd() *cobra.Command {
 				return errors.Wrap(removeErr, "failed to remove environment containers. Please remove them manually")
 			}
 
+			if err := stopExtensions(); err != nil {
+				framework.L.Warn().Msgf("failed to stop extensions: %s", err)
+			}
+
 			if allFlag {
 				stopChipIngressStackErr := stopChipIngressStack()
 				if stopChipIngressStackErr != nil {
@@ -814,6 +832,9 @@ func defaultGatewayWhitelistConfig(in *envconfig.Config, extraAllowedPorts []int
 	if in.FakeHTTP != nil {
 		ports = append(ports, in.FakeHTTP.Port)
 	}
+	if in.Extensions != nil {
+		ports = append(ports, artifactStorePort)
+	}
 
 	return gateway.WhitelistConfig{
 		ExtraAllowedPorts:   ports,
@@ -861,6 +882,9 @@ func StartCLIEnvironment(
 		BlockchainDeployers:     blockchains_sets.NewDeployerSet(testLogger, in.Infra),
 		ContractVersions:        env.ContractVersions(),
 	}
+	if in.Extensions != nil {
+		universalSetupInput.ConfigFactoryFunctions = append(universalSetupInput.ConfigFactoryFunctions, extensionsNodeConfig)
+	}
 
 	ctx, cancel := context.WithTimeout(cmdContext, 20*time.Minute)
 	defer cancel()

core/scripts/cre/environment/environment/extensions.go

@@ -0,0 +1,193 @@
+package environment
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"slices"
+	"strconv"
+	"strings"
+
+	"github.com/pelletier/go-toml/v2"
+	"github.com/pkg/errors"
+
+	"github.com/smartcontractkit/chainlink-testing-framework/framework"
+	storage "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/cre_storage_service"
+	keystone_changeset "github.com/smartcontractkit/chainlink/deployment/keystone/changeset"
+	"github.com/smartcontractkit/chainlink/system-tests/lib/cre"
+	"github.com/smartcontractkit/chainlink/system-tests/lib/cre/contracts"
+	creenv "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment"
+	envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config"
+	coretoml "github.com/smartcontractkit/chainlink/v2/core/config/toml"
+	corechainlink "github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
+)
+
+// Role addresses the extension stacks answer to on ctf, as defined by their CTF components; configs/extensions/README.md
+// lists them.
+const (
+	extensionsSourceName = "cre-extensions"
+
+	workflowSourceRole = storage.WorkflowSourceRole + ":" + storage.NodeServicePort
+	artifactAPIRole    = storage.ArtifactAPIRole + ":" + storage.NodeServicePort
+	artifactStoreRole  = storage.ArtifactStoreRole + ":" + storage.ArtifactStorePort
+)
+
+var artifactStorePort = mustPort(storage.ArtifactStorePort)
+
+// extensionsNodeConfig points workflow DON nodes at the extension roles: workflows come from cre-workflow-source and
+// artifacts from cre-artifact-api / cre-artifact-store.
+func extensionsNodeConfig(input cre.GenerateConfigsInput, configs cre.NodeIndexToConfigOverride) (cre.NodeIndexToConfigOverride, error) {
+	if !input.DonMetadata.HasFlag(cre.WorkflowDON) {
+		return configs, nil
+	}
+
+	out := make(cre.NodeIndexToConfigOverride, len(configs))
+	for idx, raw := range configs {
+		var c corechainlink.Config
+		if err := toml.Unmarshal([]byte(raw), &c); err != nil {
+			return nil, errors.Wrapf(err, "failed to unmarshal config of node %d in DON %s", idx, input.DonMetadata.Name)
+		}
+
+		c.CRE.WorkflowFetcher = &coretoml.WorkflowFetcherConfig{URL: new("")}
+		c.Capabilities.WorkflowRegistry.WorkflowStorage = coretoml.WorkflowStorage{
+			URL:                 new(artifactAPIRole),
+			TLSEnabled:          new(false),
+			ArtifactStorageHost: new(artifactStoreRole),
+		}
+		c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig = append(c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig, coretoml.AdditionalWorkflowSource{
+			URL:        new(workflowSourceRole),
+			TLSEnabled: new(false),
+			Name:       new(extensionsSourceName),
+		})
+
+		marshalled, err := toml.Marshal(c)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to marshal config of node %d in DON %s", idx, input.DonMetadata.Name)
+		}
+		out[idx] = string(marshalled)
+	}
+	return out, nil
+}
+
+// validateExtensions rejects, before the environment starts, extensions config that can't work: a non-Docker provider,
+// a missing compose file, other than one workflow DON family, or stack images that are neither local nor pullable. It
+// returns the workflow DON family the stacks serve.
+func validateExtensions(ctx context.Context, in *envconfig.Config) (string, error) {
+	if in.Infra == nil || !in.Infra.IsDocker() {
+		return "", errors.New("extensions require the docker infra provider")
+	}
+	if _, err := os.Stat(in.Extensions.StorageCompose); err != nil {
+		return "", errors.Wrap(err, "extensions.storage_compose")
+	}
+
+	var families []string
+	for _, ns := range in.NodeSets {
+		if slices.Contains(ns.DONTypes, cre.WorkflowDON) && len(ns.DonFamilies) > 0 && !slices.Contains(families, ns.DonFamilies[0]) {
+			families = append(families, ns.DonFamilies[0])
+		}
+	}
+	if len(families) != 1 {
+		return "", errors.Errorf("extensions support exactly one workflow DON family, found %v", families)
+	}
+
+	return families[0], ensureExtensionsImages(ctx, in.Extensions)
+}
+
+// ensureExtensionsImages pulls the stacks' images that aren't local, and names the ones that can't be pulled (images
+// built from the stack's repo) instead of failing later in docker compose up.
+func ensureExtensionsImages(ctx context.Context, ext *envconfig.Extensions) error {
+	cmd := docker(ctx, "compose", "-f", ext.StorageCompose, "config", "--images")
+	// the stack's required inputs; their values don't affect the images
+	cmd.Env = append(os.Environ(),
+		"CRE_LOCAL_REGISTRY_RPC_URL=placeholder",
+		"CRE_LOCAL_REGISTRY_CHAIN_SELECTOR=0",
+		"CRE_LOCAL_REGISTRY_ADDRESS=placeholder",
+		"CRE_LOCAL_DON_FAMILY=placeholder",
+	)
+	cmd.Stderr = os.Stderr
+	out, err := cmd.Output()
+	if err != nil {
+		return errors.Wrapf(err, "failed to list images in %s", ext.StorageCompose)
+	}
+
+	images := strings.Fields(string(out))
+	slices.Sort(images)
+	var missing []string
+	for _, image := range slices.Compact(images) {
+		if docker(ctx, "image", "inspect", image).Run() != nil && docker(ctx, "pull", "-q", image).Run() != nil {
+			missing = append(missing, image)
+		}
+	}
+	if len(missing) > 0 {
+		return errors.Errorf("images %v are neither local nor pullable; build them as described in the README next to %s", missing, ext.StorageCompose)
+	}
+	return nil
+}
+
+// startExtensions starts the extension stacks on the environment's network, with the registry chain, its
+// CapabilitiesRegistry and the workflow DON family as inputs.
+func startExtensions(ctx context.Context, ext *envconfig.Extensions, donFamily string, output *creenv.SetupOutput) error {
+	in, err := storageInput(ext, output, donFamily)
+	if err != nil {
+		return err
+	}
+
+	_ = storage.Stop()
+
+	out, err := storage.NewWithContext(ctx, in)
+	if err != nil {
+		return errors.Wrapf(err, "failed to start %s", ext.StorageCompose)
+	}
+	framework.L.Info().
+		Str("cre-deploy-api", out.UserService.ExternalURL).
+		Str("cre-workflow-source", out.NodeService.ExternalURL).
+		Str("cre-artifact-store", out.ArtifactStore.ExternalURL).
+		Msg("Extensions started; reachable from the host at")
+	return nil
+}
+
+func stopExtensions() error {
+	return storage.Stop()
+}
+
+func storageInput(ext *envconfig.Extensions, output *creenv.SetupOutput, donFamily string) (*storage.Input, error) {
+	creEnv := output.CreEnvironment
+	selector := creEnv.RegistryChainSelector
+
+	var rpcURL string
+	for _, bc := range creEnv.Blockchains {
+		if bc.ChainSelector() == selector && len(bc.CtfOutput().Nodes) > 0 {
+			rpcURL = bc.CtfOutput().Nodes[0].InternalHTTPUrl
+		}
+	}
+	if rpcURL == "" {
+		return nil, errors.Errorf("no RPC URL for registry chain %d", selector)
+	}
+
+	capReg := contracts.MightGetAddressFromDataStore(creEnv.CldfEnvironment.DataStore, selector,
+		keystone_changeset.CapabilitiesRegistry.String(), creEnv.ContractVersions[keystone_changeset.CapabilitiesRegistry.String()], "")
+	if capReg == nil {
+		return nil, errors.Errorf("no CapabilitiesRegistry on registry chain %d", selector)
+	}
+
+	return &storage.Input{
+		ComposeFile:           "file://" + ext.StorageCompose,
+		Network:               framework.DefaultNetworkName,
+		RegistryRPCURL:        rpcURL,
+		RegistryChainSelector: selector,
+		RegistryAddress:       capReg.Hex(),
+		DonFamily:             donFamily,
+	}, nil
+}
+
+func mustPort(port string) int {
+	p, err := strconv.Atoi(port)
+	if err != nil {
+		panic(errors.Wrapf(err, "invalid port %q", port))
+	}
+	return p
+}
+
+func docker(ctx context.Context, args ...string) *exec.Cmd {
+	return exec.CommandContext(ctx, "docker", args...)
+}

core/scripts/cre/environment/environment/extensions_test.go

@@ -0,0 +1,97 @@
+package environment
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/pelletier/go-toml/v2"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink/system-tests/lib/cre"
+	envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config"
+	"github.com/smartcontractkit/chainlink/system-tests/lib/infra"
+	corechainlink "github.com/smartcontractkit/chainlink/v2/core/services/chainlink"
+)
+
+const extensionsTestNodeConfig = `
+[CRE.WorkflowFetcher]
+URL = 'file:///home/chainlink/workflows'
+
+[[Capabilities.WorkflowRegistry.AdditionalSources]]
+URL = 'grpc-source:50051'
+Name = 'existing'
+`
+
+func TestExtensionsNodeConfig_workflowDON(t *testing.T) {
+	t.Parallel()
+
+	out, err := extensionsNodeConfig(extensionsTestInput(cre.WorkflowDON), cre.NodeIndexToConfigOverride{0: extensionsTestNodeConfig, 1: extensionsTestNodeConfig})
+	require.NoError(t, err)
+	require.Len(t, out, 2)
+
+	var c corechainlink.Config
+	require.NoError(t, toml.Unmarshal([]byte(out[1]), &c))
+	require.Empty(t, *c.CRE.WorkflowFetcher.URL)
+
+	storage := c.Capabilities.WorkflowRegistry.WorkflowStorage
+	require.Equal(t, "cre-artifact-api:50052", *storage.URL)
+	require.Equal(t, "cre-artifact-store:4566", *storage.ArtifactStorageHost)
+	require.False(t, *storage.TLSEnabled)
+
+	sources := c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig
+	require.Len(t, sources, 2)
+	require.Equal(t, "existing", *sources[0].Name)
+	require.Equal(t, "cre-workflow-source:50052", *sources[1].URL)
+	require.False(t, *sources[1].TLSEnabled)
+}
+
+func TestExtensionsNodeConfig_otherDONsUnchanged(t *testing.T) {
+	t.Parallel()
+
+	configs := cre.NodeIndexToConfigOverride{0: extensionsTestNodeConfig}
+	out, err := extensionsNodeConfig(extensionsTestInput(cre.GatewayDON), configs)
+	require.NoError(t, err)
+	require.Equal(t, configs, out)
+}
+
+func extensionsTestInput(flag cre.CapabilityFlag) cre.GenerateConfigsInput {
+	return cre.GenerateConfigsInput{DonMetadata: &cre.DonMetadata{Name: "don", Flags: []string{flag}}}
+}
+
+func TestValidateExtensions_rejectsBeforeStart(t *testing.T) {
+	t.Parallel()
+
+	compose := filepath.Join(t.TempDir(), "compose.yaml")
+	require.NoError(t, os.WriteFile(compose, []byte("services: {}\n"), 0o600))
+	workflowNodeSet := func(family string) *cre.NodeSet {
+		return &cre.NodeSet{DONTypes: []string{cre.WorkflowDON}, DonFamilies: []string{family}}
+	}
+
+	tests := []struct {
+		name string
+		in   *envconfig.Config
+		want string
+	}{
+		{"kubernetes", &envconfig.Config{
+			Infra:      &infra.Provider{Type: infra.Kubernetes},
+			Extensions: &envconfig.Extensions{StorageCompose: compose},
+		}, "docker infra provider"},
+		{"missing compose file", &envconfig.Config{
+			Infra:      &infra.Provider{Type: infra.Docker},
+			Extensions: &envconfig.Extensions{StorageCompose: filepath.Join(t.TempDir(), "missing.yaml")},
+		}, "extensions.storage_compose"},
+		{"two workflow DON families", &envconfig.Config{
+			Infra:      &infra.Provider{Type: infra.Docker},
+			NodeSets:   []*cre.NodeSet{workflowNodeSet("zone-a"), workflowNodeSet("zone-b")},
+			Extensions: &envconfig.Extensions{StorageCompose: compose},
+		}, "exactly one workflow DON family"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			_, err := validateExtensions(t.Context(), tt.in)
+			require.ErrorContains(t, err, tt.want)
+		})
+	}
+}

core/scripts/cre/environment/environment/workflow.go

@@ -126,6 +126,9 @@ func deployWorkflowCmd() *cobra.Command {
 			if resolverErr != nil {
 				return errors.Wrap(resolverErr, "failed to load local CRE state")
 			}
+			if resolver != nil && resolver.cfg.Extensions != nil {
+				return errors.New("❌ this environment reads workflows from its extensions, not the file fetcher; deploy through cre-deploy-api instead (see configs/extensions/README.md)")
+			}
 
 			defer func() {
 				metaData := map[string]any{}

core/scripts/go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20261005112317-b723176adfe8
 	github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.8
-	github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.23
+	github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.24-0.20261022155844-32538607cda4
 	github.com/smartcontractkit/chainlink-testing-framework/seth v1.51.5
 	github.com/smartcontractkit/chainlink/deployment v0.0.0-20260126202327-6be9a05f0caf
 	github.com/smartcontractkit/chainlink/system-tests/lib v0.0.0-00010101000000-000000000000