stackruntime: ApplyPlan function

This is the exported entry-point for applying a plan that was created
by an earlier call to stackruntime.Plan.

internal/stacks/stackruntime/apply.go

@@ -0,0 +1,163 @@
+package stackruntime
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/hashicorp/terraform/internal/addrs"
+	"github.com/hashicorp/terraform/internal/providers"
+	"github.com/hashicorp/terraform/internal/stacks/stackconfig"
+	"github.com/hashicorp/terraform/internal/stacks/stackruntime/internal/stackeval"
+	"github.com/hashicorp/terraform/internal/stacks/stackstate"
+	"github.com/hashicorp/terraform/internal/tfdiags"
+)
+
+// Apply performs the changes described in a previously-generated plan,
+// aiming to make the real system converge with the desired state and
+// then emit a series of patches that the caller must make to the
+// current state to represent what has changed.
+//
+// Apply does not return a result directly because it emits results in a
+// streaming fashion using channels provided in the given [ApplyResponse].
+//
+// Callers must not modify any values reachable directly or indirectly
+// through resp after passing it to this function, aside from the implicit
+// modifications to the internal state of channels caused by reading them.
+func Apply(ctx context.Context, req *ApplyRequest, resp *ApplyResponse) {
+	resp.Complete = false // We'll reset this to true only if we actually succeed
+
+	var seenAnyErrors atomic.Bool
+	outp := stackeval.ApplyOutput{
+		AnnounceAppliedChange: func(ctx context.Context, change stackstate.AppliedChange) {
+			resp.AppliedChanges <- change
+		},
+		AnnounceDiagnostics: func(ctx context.Context, diags tfdiags.Diagnostics) {
+			for _, diag := range diags {
+				if diag.Severity() == tfdiags.Error {
+					seenAnyErrors.Store(true) // never becomes false again
+				}
+				resp.Diagnostics <- diag
+			}
+		},
+	}
+
+	main, err := stackeval.ApplyPlan(
+		ctx,
+		req.Config,
+		req.RawPlan,
+		stackeval.ApplyOpts{
+			ProviderFactories: req.ProviderFactories,
+		},
+		outp,
+	)
+	if err != nil {
+		// An error here means that the apply wasn't even able to _start_,
+		// typically because the request itself was invalid. We'll announce
+		// that as a diagnostic and then halt, though if we get here then
+		// it's most likely a bug in the caller rather than end-user error.
+		resp.Diagnostics <- tfdiags.Sourceless(
+			tfdiags.Error,
+			"Invalid apply request",
+			fmt.Sprintf("Cannot begin the apply phase: %s.", err),
+		)
+		return
+	}
+
+	if !seenAnyErrors.Load() {
+		resp.Complete = true
+	}
+
+	cleanupDiags := main.DoCleanup(ctx)
+	for _, diag := range cleanupDiags {
+		// cleanup diagnostics don't stop the apply from being "complete",
+		// since this should include only transient operational errors such
+		// as failing to terminate a provider plugin.
+		resp.Diagnostics <- diag
+	}
+
+	close(resp.Diagnostics)
+	close(resp.AppliedChanges) // MUST be the last channel to close
+}
+
+// ApplyRequest represents the inputs to an [Apply] call.
+type ApplyRequest struct {
+	Config  *stackconfig.Config
+	RawPlan []*anypb.Any
+
+	ProviderFactories map[addrs.Provider]providers.Factory
+}
+
+// ApplyResponse is used by [Apply] to describe the results of applying.
+//
+// [Apply] produces streaming results throughout its execution, and so it
+// communicates with the caller by writing to provided channels during its work
+// and then modifying other fields in this structure before returning. Callers
+// MUST NOT access any non-channel fields of ApplyResponse until the
+// AppliedChanges channel has been closed to signal the completion of the
+// apply process.
+type ApplyResponse struct {
+	// [Apply] will set this field to true if the apply ran to completion
+	// without encountering any errors, or set this to false if not.
+	//
+	// A caller might react to Complete: true by creating one follow-up plan
+	// just to confirm that everything has converged and then, if so, consider
+	// all of the configuration versions that contributed to this plan to now
+	// be converged. If unsuccessful, none of the contributing configurations
+	// are known to be converged and the operator will need to decide whether
+	// to immediately try creating a new plan (if they think the error was
+	// transient) or push a new configuration update to correct the problem.
+	//
+	// If this field is false after applying is complete then it's likely that
+	// at least some of the planned side-effects already occurred, and so
+	// it's important to still handle anything that was written to the
+	// AppliedChanges channel to partially update the state with the subset
+	// of changes that were completed.
+	//
+	// The initial value of this field is ignored; there's no reason to set
+	// it to anything other than the zero value.
+	Complete bool
+
+	// AppliedChanges is the channel that will be sent each individual
+	// applied change, in no predictable order, during the apply
+	// operation.
+	//
+	// Callers MUST provide a non-nil channel and read from it from
+	// another Goroutine throughout the apply operation, or apply
+	// progress will be blocked. Callers that read slowly should provide
+	// a buffered channel to reduce the backpressure they exert on the
+	// apply process.
+	//
+	// The apply operation will close this channel before it returns.
+	// AppliedChanges is guaranteed to be the last channel to close
+	// (i.e. after Diagnostics is closed) so callers can use the close
+	// signal of this channel alone to mark that the apply process is
+	// over, but if Diagnostics is a buffered channel they must take
+	// care to deplete its buffer afterwards to avoid losing diagnostics
+	// delivered near the end of the apply process.
+	AppliedChanges chan<- stackstate.AppliedChange
+
+	// Diagnostics is the channel that will be sent any diagnostics
+	// that arise during the apply process, in no particular order.
+	//
+	// In particular note that there's no guarantee that the diagnostics
+	// for applying changes to a particular object will be emitted in close
+	// proximity to an AppliedChanges write for that same object. Diagnostics
+	// and applied changes are totally decoupled, since diagnostics might be
+	// collected up and emitted later as a large batch if the runtime
+	// needs to perform aggregate operations such as deduplication on
+	// the diagnostics before exposing them.
+	//
+	// Callers MUST provide a non-nil channel and read from it from
+	// another Goroutine throughout the plan operation, or apply
+	// progress will be blocked. Callers that read slowly should provide
+	// a buffered channel to reduce the backpressure they exert on the
+	// apply process.
+	//
+	// The apply operation will close this channel before it returns, but
+	// callers should use the close event of AppliedChanges as the definitive
+	// signal that planning is complete.
+	Diagnostics chan<- tfdiags.Diagnostic
+}

internal/stacks/stackruntime/internal/stackeval/main_apply.go

@@ -17,17 +17,28 @@ import (
 // ApplyPlan internally instantiates a [Main] configured to apply the given
 // raw plan, and then visits all of the relevant objects to collect up any
 // diagnostics they emit while evaluating in terms of the change results.
-func ApplyPlan(ctx context.Context, config *stackconfig.Config, rawPlan []*anypb.Any, opts ApplyOpts, outp ApplyOutput) error {
+//
+// If the error result is non-nil then that means the apply process didn't
+// even begin, because the given arguments were invalid. If the arguments
+// are valid enough to start the apply process then the error will always
+// be nil and any problems along the way will be reported as diagnostics
+// through the [ApplyOutput] object.
+//
+// Returns the [Main] object that was used to track state during the process.
+// Callers must call [Main.DoCleanup] on that object once they've finished
+// with it to avoid leaking non-memory resources such as goroutines and
+// provider plugin processes.
+func ApplyPlan(ctx context.Context, config *stackconfig.Config, rawPlan []*anypb.Any, opts ApplyOpts, outp ApplyOutput) (*Main, error) {
 	plan, err := stackplan.LoadFromProto(rawPlan)
 	if err != nil {
-		return fmt.Errorf("invalid raw plan: %w", err)
+		return nil, fmt.Errorf("invalid raw plan: %w", err)
 	}
 	if !plan.Applyable {
 		// We should not get here because a caller should not ask us to try
 		// to apply a plan that wasn't marked as applyable, but we'll check
 		// it anyway just to be robust in case there's a bug further up
 		// the call stack.
-		return fmt.Errorf("plan is not applyable")
+		return nil, fmt.Errorf("plan is not applyable")
 	}
 
 	// We'll register all of the changes we intend to make up front, so we
@@ -152,7 +163,7 @@ func ApplyPlan(ctx context.Context, config *stackconfig.Config, rawPlan []*anypb
 		outp.AnnounceDiagnostics(ctx, diags)
 	}
 
-	return nil
+	return main, nil
 }
 
 type ApplyOutput struct {