[engine] Clear pending operations with refresh.

CHANGELOG_PENDING.md

@@ -1,5 +1,8 @@
 ### Improvements
 
+- Clear pending operations during `pulumi refresh` or `pulumi up -r`.
+  [#8435](https://github.com/pulumi/pulumi/pull/8435)
+
 ### Bug Fixes
 
 - [codegen/go] - Fix Go SDK function output to check for errors

pkg/backend/snapshot.go

@@ -600,6 +600,15 @@ func (sm *SnapshotManager) snap() *deploy.Snapshot {
 		}
 	}
 
+	// Track pending create operations from the base snapshot
+	// and propagate them to the new snapshot: we don't want to clear pending CREATE operations
+	// because these must require user intervention to be cleared or resolved.
+	for _, pendingOperation := range sm.baseSnapshot.PendingOperations {
+		if pendingOperation.Type == resource.OperationTypeCreating {
+			operations = append(operations, pendingOperation)
+		}
+	}
+
 	manifest := deploy.Manifest{
 		Time:    time.Now(),
 		Version: version.Version,

pkg/engine/journal.go

@@ -111,6 +111,17 @@ func (entries JournalEntries) Snap(base *deploy.Snapshot) *deploy.Snapshot {
 		}
 	}
 
+	if base != nil {
+		// Track pending create operations from the base snapshot
+		// and propagate them to the new snapshot: we don't want to clear pending CREATE operations
+		// because these must require user intervention to be cleared or resolved.
+		for _, pendingOperation := range base.PendingOperations {
+			if pendingOperation.Type == resource.OperationTypeCreating {
+				operations = append(operations, pendingOperation)
+			}
+		}
+	}
+
 	// If we have a base snapshot, copy over its secrets manager.
 	var secretsManager secrets.Manager
 	if base != nil {

pkg/engine/lifeycletest/pulumi_test.go

@@ -557,6 +557,75 @@ func TestPreviewWithPendingOperations(t *testing.T) {
 	assert.EqualError(t, res.Error(), deploy.PlanPendingOperationsError{}.Error())
 }
 
+// Tests that a refresh works for a stack with pending operations.
+func TestRefreshWithPendingOperations(t *testing.T) {
+	t.Parallel()
+
+	p := &TestPlan{}
+
+	const resType = "pkgA:m:typA"
+	urnA := p.NewURN(resType, "resA", "")
+
+	newResource := func(urn resource.URN, id resource.ID, delete bool, dependencies ...resource.URN) *resource.State {
+		return &resource.State{
+			Type:         urn.Type(),
+			URN:          urn,
+			Custom:       true,
+			Delete:       delete,
+			ID:           id,
+			Inputs:       resource.PropertyMap{},
+			Outputs:      resource.PropertyMap{},
+			Dependencies: dependencies,
+		}
+	}
+
+	old := &deploy.Snapshot{
+		PendingOperations: []resource.Operation{{
+			Resource: newResource(urnA, "0", false),
+			Type:     resource.OperationTypeUpdating,
+		}},
+		Resources: []*resource.State{
+			newResource(urnA, "0", false),
+		},
+	}
+
+	loaders := []*deploytest.ProviderLoader{
+		deploytest.NewProviderLoader("pkgA", semver.MustParse("1.0.0"), func() (plugin.Provider, error) {
+			return &deploytest.Provider{}, nil
+		}),
+	}
+
+	program := deploytest.NewLanguageRuntime(func(_ plugin.RunInfo, monitor *deploytest.ResourceMonitor) error {
+		_, _, _, err := monitor.RegisterResource("pkgA:m:typA", "resA", true)
+		assert.NoError(t, err)
+		return nil
+	})
+
+	op := TestOp(Update)
+	options := UpdateOptions{Host: deploytest.NewPluginHost(nil, nil, program, loaders...)}
+	project, target := p.GetProject(), p.GetTarget(t, old)
+
+	// Without refreshing, an update should fail.
+	_, res := op.Run(project, target, options, false, nil, nil)
+	assertIsErrorOrBailResult(t, res)
+	assert.EqualError(t, res.Error(), deploy.PlanPendingOperationsError{}.Error())
+
+	// With a refresh, the update should succeed.
+	withRefresh := options
+	withRefresh.Refresh = true
+	new, res := op.Run(project, target, withRefresh, false, nil, nil)
+	assert.Nil(t, res)
+	assert.Len(t, new.PendingOperations, 0)
+
+	// Similarly, the update should succeed if performed after a separate refresh.
+	new, res = TestOp(Refresh).Run(project, target, options, false, nil, nil)
+	assert.Nil(t, res)
+	assert.Len(t, new.PendingOperations, 0)
+
+	_, res = op.Run(project, p.GetTarget(t, new), options, false, nil, nil)
+	assert.Nil(t, res)
+}
+
 // Tests that a failed partial update causes the engine to persist the resource's old inputs and new outputs.
 func TestUpdatePartialFailure(t *testing.T) {
 	t.Parallel()

pkg/resource/deploy/deployment.go

@@ -305,10 +305,6 @@ func buildResourceMap(prev *Snapshot, preview bool) ([]*resource.State, map[reso
 		return nil, olds, nil
 	}
 
-	if prev.PendingOperations != nil && !preview {
-		return nil, nil, PlanPendingOperationsError{prev.PendingOperations}
-	}
-
 	for _, oldres := range prev.Resources {
 		// Ignore resources that are pending deletion; these should not be recorded in the LUT.
 		if oldres.Delete {

pkg/resource/deploy/deployment_executor.go

@@ -146,6 +146,8 @@ func (ex *deploymentExecutor) Execute(callerCtx context.Context, opts Options, p
 		if opts.RefreshOnly {
 			return nil, nil
 		}
+	} else if ex.deployment.prev != nil && len(ex.deployment.prev.PendingOperations) != 0 && !preview {
+		return nil, result.FromError(PlanPendingOperationsError{ex.deployment.prev.PendingOperations})
 	}
 
 	// The set of -t targets provided on the command line.  'nil' means 'update everything'.

pkg/resource/deploy/deployment_test.go

@@ -45,16 +45,5 @@ func TestPendingOperationsDeployment(t *testing.T) {
 	})
 
 	_, err := NewDeployment(&plugin.Context{}, &Target{}, snap, nil, &fixedSource{}, nil, false, nil)
-	if !assert.Error(t, err) {
-		t.FailNow()
-	}
-
-	invalidErr, ok := err.(PlanPendingOperationsError)
-	if !assert.True(t, ok) {
-		t.FailNow()
-	}
-
-	assert.Len(t, invalidErr.Operations, 1)
-	assert.Equal(t, resourceB.URN, invalidErr.Operations[0].Resource.URN)
-	assert.Equal(t, resource.OperationTypeCreating, invalidErr.Operations[0].Type)
+	assert.NoError(t, err)
 }