Merge #11151

11151: Prevent concurrent writes to the component providers map r=iwahbe a=iwahbe

<!--- 
Thanks so much for your contribution! If this is your first time contributing, please ensure that you have read the [CONTRIBUTING](https://github.com/pulumi/pulumi/blob/master/CONTRIBUTING.md) documentation.
-->

# Description

<!--- Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. -->

Fixes # (issue)

## Checklist

<!--- Please provide details if the checkbox below is to be left unchecked. -->
- [ ] I have added tests that prove my fix is effective or that my feature works
<!--- 
User-facing changes require a CHANGELOG entry.
-->
- [ ] I have run `make changelog` and committed the `changelog/pending/<file>` documenting my change
<!--
If the change(s) in this PR is a modification of an existing call to the Pulumi Service,
then the service should honor older versions of the CLI where this change would not exist.
You must then bump the API version in /pkg/backend/httpstate/client/api.go, as well as add
it to the service.
-->
- [ ] Yes, there are changes in this PR that warrants bumping the Pulumi Service API version
  <!-- `@Pulumi` employees: If yes, you must submit corresponding changes in the service repo. -->


Co-authored-by: Ian Wahbe <ian@wahbe.com>

changelog/pending/20221025--engine--prevent-concurrent-read-writes-to-the-component-providers-map.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: fix
+  scope: engine
+  description: Prevent concurrent read/writes to the component providers map.

pkg/resource/deploy/source_eval.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/blang/semver"
@@ -477,18 +478,19 @@ func (d *defaultProviders) getDefaultProviderRef(req providers.ProviderRequest)
 // resmon implements the pulumirpc.ResourceMonitor interface and acts as the gateway between a language runtime's
 // evaluation of a program and the internal resource planning and deployment logic.
 type resmon struct {
-	diagostics                 diag.Sink                          // logger for user-facing messages
-	providers                  ProviderSource                     // the provider source itself.
-	componentResourceProviders map[resource.URN]map[string]string // which providers component resources used
-	defaultProviders           *defaultProviders                  // the default provider manager.
-	constructInfo              plugin.ConstructInfo               // information for construct and call calls.
-	regChan                    chan *registerResourceEvent        // the channel to send resource registrations to.
-	regOutChan                 chan *registerResourceOutputsEvent // the channel to send resource output registrations to.
-	regReadChan                chan *readResourceEvent            // the channel to send resource reads to.
-	cancel                     chan bool                          // a channel that can cancel the server.
-	done                       chan error                         // a channel that resolves when the server completes.
-	disableResourceReferences  bool                               // true if resource references are disabled.
-	disableOutputValues        bool                               // true if output values are disabled.
+	diagostics                diag.Sink                          // logger for user-facing messages
+	providers                 ProviderSource                     // the provider source itself.
+	componentProviders        map[resource.URN]map[string]string // which providers component resources used
+	componentProvidersLock    sync.Mutex                         // which locks the componentProviders map
+	defaultProviders          *defaultProviders                  // the default provider manager.
+	constructInfo             plugin.ConstructInfo               // information for construct and call calls.
+	regChan                   chan *registerResourceEvent        // the channel to send resource registrations to.
+	regOutChan                chan *registerResourceOutputsEvent // the channel to send resource output registrations to.
+	regReadChan               chan *readResourceEvent            // the channel to send resource reads to.
+	cancel                    chan bool                          // a channel that can cancel the server.
+	done                      chan error                         // a channel that resolves when the server completes.
+	disableResourceReferences bool                               // true if resource references are disabled.
+	disableOutputValues       bool                               // true if output values are disabled.
 }
 
 var _ SourceResourceMonitor = (*resmon)(nil)
@@ -513,16 +515,16 @@ func newResourceMonitor(src *evalSource, provs ProviderSource, regChan chan *reg
 
 	// New up an engine RPC server.
 	resmon := &resmon{
-		diagostics:                 src.plugctx.Diag,
-		providers:                  provs,
-		defaultProviders:           d,
-		componentResourceProviders: map[resource.URN]map[string]string{},
-		regChan:                    regChan,
-		regOutChan:                 regOutChan,
-		regReadChan:                regReadChan,
-		cancel:                     cancel,
-		disableResourceReferences:  opts.DisableResourceReferences,
-		disableOutputValues:        opts.DisableOutputValues,
+		diagostics:                src.plugctx.Diag,
+		providers:                 provs,
+		defaultProviders:          d,
+		componentProviders:        map[resource.URN]map[string]string{},
+		regChan:                   regChan,
+		regOutChan:                regOutChan,
+		regReadChan:               regReadChan,
+		cancel:                    cancel,
+		disableResourceReferences: opts.DisableResourceReferences,
+		disableOutputValues:       opts.DisableOutputValues,
 	}
 
 	// Fire up a gRPC server and start listening for incomings.
@@ -969,17 +971,22 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 
 	// We handle updating the providers map to include the providers field of the parent if
 	// both the current resource and its parent is a component resource.
-	if parentsProviders, parentIsComponent := rm.componentResourceProviders[parent]; !custom &&
-		parent != "" && parentIsComponent {
-		for k, v := range parentsProviders {
-			if req.Providers == nil {
-				req.Providers = map[string]string{}
-			}
-			if _, ok := req.Providers[k]; !ok {
-				req.Providers[k] = v
+	func() {
+		// Function exists to scope the lock
+		rm.componentProvidersLock.Lock()
+		defer rm.componentProvidersLock.Unlock()
+		if parentsProviders, parentIsComponent := rm.componentProviders[parent]; !custom &&
+			parent != "" && parentIsComponent {
+			for k, v := range parentsProviders {
+				if req.Providers == nil {
+					req.Providers = map[string]string{}
+				}
+				if _, ok := req.Providers[k]; !ok {
+					req.Providers[k] = v
+				}
 			}
 		}
-	}
+	}()
 
 	label := fmt.Sprintf("ResourceMonitor.RegisterResource(%s,%s)", t, name)
 
@@ -1201,7 +1208,11 @@ func (rm *resmon) RegisterResource(ctx context.Context,
 	}
 
 	if !custom && result != nil && result.State != nil && result.State.URN != "" {
-		rm.componentResourceProviders[result.State.URN] = req.GetProviders()
+		func() {
+			rm.componentProvidersLock.Lock()
+			defer rm.componentProvidersLock.Unlock()
+			rm.componentProviders[result.State.URN] = req.GetProviders()
+		}()
 	}
 
 	// Filter out partially-known values if the requestor does not support them.