fix(test/runtime): pass input down instead of using env var

Author: graham-chainlink

engine/cld/changeset/common.go

@@ -77,6 +77,34 @@ type TypedJSON struct {
 	ChainOverrides []uint64        `json:"chainOverrides"` // Optional field for chain overrides
 }
 
+func parseTypedInput(inputStr string) (TypedJSON, error) {
+	if inputStr == "" {
+		return TypedJSON{}, errors.New("input is empty")
+	}
+
+	var inputObject TypedJSON
+	if err := json.Unmarshal([]byte(inputStr), &inputObject); err != nil {
+		return TypedJSON{}, fmt.Errorf("JSON must be in JSON format with 'payload' fields: %w", err)
+	}
+	if len(inputObject.Payload) == 0 {
+		return TypedJSON{}, errors.New("'payload' field is required")
+	}
+
+	return inputObject, nil
+}
+
+func decodePayload[C any](payload json.RawMessage) (C, error) {
+	var config C
+
+	payloadDecoder := json.NewDecoder(strings.NewReader(string(payload)))
+	payloadDecoder.DisallowUnknownFields()
+	if err := payloadDecoder.Decode(&config); err != nil {
+		return config, fmt.Errorf("failed to unmarshal payload: %w", err)
+	}
+
+	return config, nil
+}
+
 // WithJSON returns a fully configured changeset, which pairs a [fdeployment.ChangeSet] with its configuration based
 // a JSON input. It also allows extensions, such as a PostProcessing function.
 // InputStr must be a JSON object with a "payload" field that contains the actual input data for a Durable Pipeline.
@@ -92,30 +120,13 @@ type TypedJSON struct {
 // Note: Prefer WithEnvInput for durable_pipelines.go
 func (f WrappedChangeSet[C]) WithJSON(_ C, inputStr string) ConfiguredChangeSet {
 	return ChangeSetImpl[C]{changeset: f, configProvider: func() (C, error) {
-		var config C
-
-		if inputStr == "" {
-			return config, errors.New("input is empty")
-		}
-
-		var inputObject TypedJSON
-		if err := json.Unmarshal([]byte(inputStr), &inputObject); err != nil {
-			return config, fmt.Errorf("JSON must be in JSON format with 'payload' fields: %w", err)
-		}
-
-		// If payload is null, decode it as null (which will give zero value)
-		// If payload is missing, return an error
-		if len(inputObject.Payload) == 0 {
-			return config, errors.New("'payload' field is required")
-		}
-
-		payloadDecoder := json.NewDecoder(strings.NewReader(string(inputObject.Payload)))
-		payloadDecoder.DisallowUnknownFields()
-		if err := payloadDecoder.Decode(&config); err != nil {
-			return config, fmt.Errorf("failed to unmarshal payload: %w", err)
+		inputObject, err := parseTypedInput(inputStr)
+		if err != nil {
+			var zero C
+			return zero, err
 		}
 
-		return config, nil
+		return decodePayload[C](inputObject.Payload)
 	},
 		inputChainOverrides: func() ([]uint64, error) {
 			return loadInputChainOverrides(inputStr)
@@ -151,41 +162,36 @@ func (f WrappedChangeSet[C]) WithEnvInput(opts ...EnvInputOption[C]) ConfiguredC
 
 	inputStr := os.Getenv("DURABLE_PIPELINE_INPUT")
 
-	return ChangeSetImpl[C]{changeset: f, configProvider: func() (C, error) {
-		var config C
-
-		if inputStr == "" {
-			return config, errors.New("input is empty")
-		}
-
-		var inputObject TypedJSON
-		if err := json.Unmarshal([]byte(inputStr), &inputObject); err != nil {
-			return config, fmt.Errorf("JSON must be in JSON format with 'payload' fields: %w", err)
-		}
+	providerFromInput := func(rawInput string) (C, error) {
+		var zero C
 
-		// If payload is null, decode it as null (which will give zero value)
-		// If payload is missing, return an error
-		if len(inputObject.Payload) == 0 {
-			return config, errors.New("'payload' field is required")
+		inputObject, err := parseTypedInput(rawInput)
+		if err != nil {
+			return zero, err
 		}
 
-		payloadDecoder := json.NewDecoder(strings.NewReader(string(inputObject.Payload)))
-		payloadDecoder.DisallowUnknownFields()
-		if err := payloadDecoder.Decode(&config); err != nil {
-			return config, fmt.Errorf("failed to unmarshal payload: %w", err)
+		config, err := decodePayload[C](inputObject.Payload)
+		if err != nil {
+			return zero, err
 		}
 
 		if options.inputModifier != nil {
-			conf, err := options.inputModifier(config)
-			if err != nil {
-				return conf, fmt.Errorf("failed to apply input modifier: %w", err)
+			conf, modifierErr := options.inputModifier(config)
+			if modifierErr != nil {
+				return conf, fmt.Errorf("failed to apply input modifier: %w", modifierErr)
 			}
 
 			return conf, nil
 		}
 
 		return config, nil
-	},
+	}
+
+	return ChangeSetImpl[C]{changeset: f,
+		configProvider: func() (C, error) {
+			return providerFromInput(inputStr)
+		},
+		configProviderWithInput: providerFromInput,
 		inputChainOverrides: func() ([]uint64, error) {
 			return loadInputChainOverrides(inputStr)
 		},
@@ -221,21 +227,17 @@ func (f WrappedChangeSet[C]) WithConfigResolver(resolver fresolvers.ConfigResolv
 	// Read input from environment variable
 	inputStr := os.Getenv("DURABLE_PIPELINE_INPUT")
 
-	configProvider := func() (C, error) {
+	configProviderFromInput := func(rawInput string) (C, error) {
 		var zero C
 
-		if inputStr == "" {
+		if rawInput == "" {
 			return zero, errors.New("input is empty")
 		}
 
-		// Parse JSON input
 		var inputObject TypedJSON
-		if err := json.Unmarshal([]byte(inputStr), &inputObject); err != nil {
+		if err := json.Unmarshal([]byte(rawInput), &inputObject); err != nil {
 			return zero, fmt.Errorf("failed to parse resolver input as JSON: %w", err)
 		}
-
-		// If payload is null, pass it to the resolver (which will receive null)
-		// If payload field is missing, return an error
 		if len(inputObject.Payload) == 0 {
 			return zero, errors.New("'payload' field is required")
 		}
@@ -249,8 +251,12 @@ func (f WrappedChangeSet[C]) WithConfigResolver(resolver fresolvers.ConfigResolv
 		return typedConfig, nil
 	}
 
-	return ChangeSetImpl[C]{changeset: f, configProvider: configProvider,
-		ConfigResolver: resolver,
+	return ChangeSetImpl[C]{changeset: f,
+		configProvider: func() (C, error) {
+			return configProviderFromInput(inputStr)
+		},
+		configProviderWithInput: configProviderFromInput,
+		ConfigResolver:          resolver,
 		inputChainOverrides: func() ([]uint64, error) {
 			return loadInputChainOverrides(inputStr)
 		},
@@ -260,9 +266,10 @@ func (f WrappedChangeSet[C]) WithConfigResolver(resolver fresolvers.ConfigResolv
 var _ ConfiguredChangeSet = ChangeSetImpl[any]{}
 
 type ChangeSetImpl[C any] struct {
-	changeset           WrappedChangeSet[C]
-	configProvider      func() (C, error)
-	inputChainOverrides func() ([]uint64, error)
+	changeset               WrappedChangeSet[C]
+	configProvider          func() (C, error)
+	configProviderWithInput func(inputStr string) (C, error)
+	inputChainOverrides     func() ([]uint64, error)
 
 	// Present only when the changeset was wired with
 	// Configure(...).WithConfigResolver(...)
@@ -287,6 +294,22 @@ func (ccs ChangeSetImpl[C]) Apply(env fdeployment.Environment) (fdeployment.Chan
 	return ccs.changeset.operation.Apply(env, c)
 }
 
+func (ccs ChangeSetImpl[C]) applyWithInput(env fdeployment.Environment, inputStr string) (fdeployment.ChangesetOutput, error) {
+	if ccs.configProviderWithInput == nil {
+		return ccs.Apply(env)
+	}
+
+	c, err := ccs.configProviderWithInput(inputStr)
+	if err != nil {
+		return fdeployment.ChangesetOutput{}, err
+	}
+	if err := ccs.changeset.operation.VerifyPreconditions(env, c); err != nil {
+		return fdeployment.ChangesetOutput{}, err
+	}
+
+	return ccs.changeset.operation.Apply(env, c)
+}
+
 func (ccs ChangeSetImpl[C]) Configurations() (Configurations, error) {
 	var chainOverrides []uint64
 	var err error

engine/cld/changeset/postprocess.go

@@ -35,6 +35,18 @@ func (ccs PostProcessingChangeSetImpl[C]) Apply(env fdeployment.Environment) (fd
 	return ccs.postProcessor(env, output)
 }
 
+func (ccs PostProcessingChangeSetImpl[C]) applyWithInput(
+	env fdeployment.Environment, inputStr string,
+) (fdeployment.ChangesetOutput, error) {
+	env.Logger.Debugf("Post-processing ChangesetOutput from %T", ccs.changeset.changeset.operation)
+	output, err := ccs.changeset.applyWithInput(env, inputStr)
+	if err != nil {
+		return output, err
+	}
+
+	return ccs.postProcessor(env, output)
+}
+
 func (ccs PostProcessingChangeSetImpl[C]) Configurations() (Configurations, error) {
 	return ccs.changeset.Configurations()
 }

engine/cld/changeset/registry.go

@@ -85,6 +85,11 @@ type hookCarrier interface {
 	getPostHooks() []PostHook
 }
 
+// inputAwareChangeSet supports explicit per-apply input injection.
+type inputAwareChangeSet interface {
+	applyWithInput(env fdeployment.Environment, inputStr string) (fdeployment.ChangesetOutput, error)
+}
+
 // newRegistryEntry creates a new registry entry for a changeset.
 func newRegistryEntry(c ChangeSet, opts ChangesetConfig) registryEntry {
 	entry := registryEntry{changeset: c, options: opts}
@@ -172,6 +177,21 @@ func (r *ChangesetsRegistry) AddGlobalPostHooks(hooks ...PostHook) {
 // a failed Apply are logged but never mask the Apply error.
 func (r *ChangesetsRegistry) Apply(
 	key string, e fdeployment.Environment,
+) (fdeployment.ChangesetOutput, error) {
+	return r.applyWithInput(key, e, "")
+}
+
+// ApplyWithInput applies a changeset with explicit input JSON for this apply invocation.
+// Changesets configured to read durable pipeline input from environment variables
+// consume this value directly when they support input-aware execution.
+func (r *ChangesetsRegistry) ApplyWithInput(
+	key string, e fdeployment.Environment, inputStr string,
+) (fdeployment.ChangesetOutput, error) {
+	return r.applyWithInput(key, e, inputStr)
+}
+
+func (r *ChangesetsRegistry) applyWithInput(
+	key string, e fdeployment.Environment, inputStr string,
 ) (fdeployment.ChangesetOutput, error) {
 	entry, globalPre, globalPost, err := r.getApplySnapshot(key)
 	if err != nil {
@@ -204,7 +224,13 @@ func (r *ChangesetsRegistry) Apply(
 		}
 	}
 
-	output, applyErr := entry.changeset.Apply(e)
+	var output fdeployment.ChangesetOutput
+	var applyErr error
+	if inputAware, ok := entry.changeset.(inputAwareChangeSet); ok {
+		output, applyErr = inputAware.applyWithInput(e, inputStr)
+	} else {
+		output, applyErr = entry.changeset.Apply(e)
+	}
 
 	postParams := PostHookParams{
 		Env:          hookEnv,

engine/cld/changeset/registry_test.go

@@ -148,6 +148,63 @@ func Test_Changesets_Apply(t *testing.T) {
 	}
 }
 
+//nolint:paralleltest // Uses process environment for fallback behavior assertions.
+func Test_Changesets_ApplyWithInput_WithEnvConfiguredChangeset(t *testing.T) {
+	type inputConfig struct {
+		Value int `json:"value"`
+	}
+
+	t.Setenv("DURABLE_PIPELINE_INPUT", `{"payload":{"value":999}}`)
+
+	var received int
+	cs := fdeployment.CreateChangeSet(
+		func(_ fdeployment.Environment, cfg inputConfig) (fdeployment.ChangesetOutput, error) {
+			received = cfg.Value
+			return fdeployment.ChangesetOutput{}, nil
+		},
+		func(_ fdeployment.Environment, _ inputConfig) error { return nil },
+	)
+
+	r := NewChangesetsRegistry()
+	r.Add("0001_test", Configure(cs).WithEnvInput())
+
+	_, err := r.ApplyWithInput("0001_test", fdeployment.Environment{}, `{"payload":{"value":1}}`)
+	require.NoError(t, err)
+	require.Equal(t, 1, received)
+}
+
+//nolint:paralleltest // Uses process environment for fallback behavior assertions.
+func Test_Changesets_ApplyWithInput_WithResolverConfiguredChangeset(t *testing.T) {
+	type resolverInput struct {
+		Base int `json:"base"`
+	}
+	type resolverOutput struct {
+		Value int `json:"value"`
+	}
+
+	t.Setenv("DURABLE_PIPELINE_INPUT", `{"payload":{"base":100}}`)
+
+	resolver := func(input resolverInput) (resolverOutput, error) {
+		return resolverOutput{Value: input.Base + 10}, nil
+	}
+
+	var received int
+	cs := fdeployment.CreateChangeSet(
+		func(_ fdeployment.Environment, cfg resolverOutput) (fdeployment.ChangesetOutput, error) {
+			received = cfg.Value
+			return fdeployment.ChangesetOutput{}, nil
+		},
+		func(_ fdeployment.Environment, _ resolverOutput) error { return nil },
+	)
+
+	r := NewChangesetsRegistry()
+	r.Add("0001_test", Configure(cs).WithConfigResolver(resolver))
+
+	_, err := r.ApplyWithInput("0001_test", fdeployment.Environment{}, `{"payload":{"base":7}}`)
+	require.NoError(t, err)
+	require.Equal(t, 17, received)
+}
+
 func Test_Changesets_Add(t *testing.T) {
 	t.Parallel()
 

engine/test/runtime/runtime_registered_changesets.go

@@ -3,7 +3,6 @@ package runtime
 import (
 	"errors"
 	"fmt"
-	"os"
 
 	"github.com/segmentio/ksuid"
 
@@ -14,14 +13,10 @@ import (
 // ExecRegisteredChangesetsFromYAML executes registered changesets from durable-pipeline YAML input.
 //
 // For each changeset entry in YAML order:
-//  1. Set DURABLE_PIPELINE_INPUT from that entry's payload/chainOverrides.
+//  1. Build per-entry durable-pipeline JSON input from payload/chainOverrides.
 //  2. Create and initialize a fresh registry provider.
-//  3. Apply the named changeset against the current runtime environment.
+//  3. Apply the named changeset against the current runtime environment with explicit input.
 //  4. Merge output into runtime state and regenerate environment for the next step.
-//
-// Do not run this in parallel. This is not thread-safe. It temporarily mutates the process-wide
-// DURABLE_PIPELINE_INPUT environment variable while applying each changeset.
-// Once we move the reliance on the environment variable in the implementation, we can remove this restriction.
 func (r *Runtime) ExecRegisteredChangesetsFromYAML(
 	providerFactory func() changeset.RegistryProvider,
 	inputYAML []byte,
@@ -46,23 +41,11 @@ func (r *Runtime) ExecRegisteredChangesetsFromYAML(
 		return fmt.Errorf("input file %s has empty 'changesets' array", "runtime-input.yaml")
 	}
 
-	oldInput, hadInput := os.LookupEnv("DURABLE_PIPELINE_INPUT")
-	defer func() {
-		if hadInput {
-			_ = os.Setenv("DURABLE_PIPELINE_INPUT", oldInput)
-		} else {
-			_ = os.Unsetenv("DURABLE_PIPELINE_INPUT")
-		}
-	}()
-
 	for _, cs := range ordered {
 		inputJSON, err := durablepipeline.BuildChangesetInputJSON(cs.Name, cs.Data)
 		if err != nil {
 			return fmt.Errorf("failed to build input for changeset %q in input file %s: %w", cs.Name, "runtime-input.yaml", err)
 		}
-		if setEnvErr := os.Setenv("DURABLE_PIPELINE_INPUT", inputJSON); setEnvErr != nil {
-			return fmt.Errorf("failed to set DURABLE_PIPELINE_INPUT environment variable: %w", setEnvErr)
-		}
 
 		provider := providerFactory()
 		if provider == nil {
@@ -72,7 +55,7 @@ func (r *Runtime) ExecRegisteredChangesetsFromYAML(
 			return fmt.Errorf("failed to init registry provider: %w", initErr)
 		}
 
-		out, err := provider.Registry().Apply(cs.Name, r.currentEnv)
+		out, err := provider.Registry().ApplyWithInput(cs.Name, r.currentEnv, inputJSON)
 		if err != nil {
 			return err
 		}