Pass state back to the engine if Apply encountered an error (#2695)

In the SDKv2 bridge under PlanResourceChange we are not passing any
state we receive during TF Apply back to the engine if we also received
an error. This causes us to incorrectly miss any resources which were
created but encountered errors during the creation process. The engine
should see these as `ResourceInitError`, which allows the engine to
attempt to update the partially created resource on the next `up`.

This PR fixes the issue by passing the state down to the engine in the
case when we receive an error and a non-nil state from TF during Apply.

related to pulumi/pulumi-gcp#2700
related to pulumi/pulumi-aws#4759

fixes #2696

pkg/tests/schema_pulumi_test.go

@@ -189,3 +189,54 @@ func TestMakeTerraformResultNilVsEmptyMap(t *testing.T) {
 		assert.True(t, props["test"].DeepEquals(emptyMap))
 	})
 }
+
+func TestResourceInitFailure(t *testing.T) {
+	t.Parallel()
+
+	resMap := map[string]*schema.Resource{
+		"prov_test": {
+			Schema: map[string]*schema.Schema{
+				"test": {
+					Type:     schema.TypeString,
+					Required: true,
+				},
+			},
+			CreateContext: func(ctx context.Context, rd *schema.ResourceData, i interface{}) diag.Diagnostics {
+				rd.SetId("1")
+				return diag.Errorf("INIT TEST ERROR")
+			},
+		},
+	}
+	prov := &schema.Provider{ResourcesMap: resMap}
+	bridgedProvider := pulcheck.BridgedProvider(t, "prov", prov)
+
+	pt := pulcheck.PulCheck(t, bridgedProvider, `
+name: test
+runtime: yaml
+resources:
+  mainRes:
+    type: prov:index:Test
+	properties:
+	  test: "hello"
+`)
+
+	_, err := pt.CurrentStack().Up(pt.Context())
+	require.Error(t, err)
+	require.ErrorContains(t, err, "INIT TEST ERROR")
+
+	stack := pt.ExportStack(t)
+
+	data, err := stack.Deployment.MarshalJSON()
+	require.NoError(t, err)
+
+	var stateMap map[string]interface{}
+	err = json.Unmarshal(data, &stateMap)
+	require.NoError(t, err)
+
+	resourcesList := stateMap["resources"].([]interface{})
+	require.Len(t, resourcesList, 3)
+	mainResState := resourcesList[2].(map[string]interface{}) // stack, provider, resource
+	initErrors := mainResState["initErrors"].([]interface{})
+	require.Len(t, initErrors, 1)
+	require.Contains(t, initErrors[0], "INIT TEST ERROR")
+}

pkg/tfshim/sdk-v2/provider2.go

@@ -347,15 +347,7 @@ func (p *planResourceChangeImpl) Apply(
 		maps.Copy(priv, diff.v2InstanceDiff.tf.Meta)
 	}
 
-	resp, err := p.server.ApplyResourceChange(ctx, t, ty, cfg, st, pl, priv, meta)
-	if err != nil {
-		return nil, err
-	}
-	return &v2InstanceState2{
-		resourceType: t,
-		stateValue:   resp.stateValue,
-		meta:         resp.meta,
-	}, nil
+	return p.server.ApplyResourceChange(ctx, t, ty, cfg, st, pl, priv, meta)
 }
 
 // This method is called to service `pulumi refresh` requests and maps naturally to the TF
@@ -662,25 +654,28 @@ func (s *grpcServer) ApplyResourceChange(
 		}
 		req.ProviderMeta = &tfprotov5.DynamicValue{MsgPack: providerMetaVal}
 	}
-	resp, err := s.gserver.ApplyResourceChange(ctx, req)
-	if err := handleDiagnostics(ctx, resp.Diagnostics, err); err != nil {
-		return nil, err
-	}
-	newState, err := msgpack.Unmarshal(resp.NewState.MsgPack, ty)
-	if err != nil {
-		return nil, err
-	}
+	resp, applyErr := s.gserver.ApplyResourceChange(ctx, req)
+	newState := cty.Value{}
 	var meta map[string]interface{}
-	if resp.Private != nil {
-		if err := json.Unmarshal(resp.Private, &meta); err != nil {
-			return nil, err
+	if resp != nil {
+		if resp.NewState != nil {
+			newState, err = msgpack.Unmarshal(resp.NewState.MsgPack, ty)
+			if err != nil {
+				return nil, err
+			}
+		}
+		if resp.Private != nil {
+			if err := json.Unmarshal(resp.Private, &meta); err != nil {
+				return nil, err
+			}
 		}
 	}
+	returnErr := handleDiagnostics(ctx, resp.Diagnostics, applyErr)
 	return &v2InstanceState2{
 		resourceType: typeName,
 		stateValue:   newState,
 		meta:         meta,
-	}, nil
+	}, returnErr
 }
 
 func (s *grpcServer) ReadResource(