[release-1.12] [BugFix] WasmPlugin: Call forwardToEnvoy in the same t…

…hread (#39176)

* Call forwardToEnvoy in the same thread

* Update pkg/istio-agent/xds_proxy.go

Co-authored-by: John Howard <howardjohn@google.com>

Co-authored-by: Ingwon Song <igsong@google.com>
Co-authored-by: Ingwon Song <102102227+ingwonsong@users.noreply.github.com>
Co-authored-by: John Howard <howardjohn@google.com>

pkg/istio-agent/xds_proxy.go

@@ -473,6 +473,7 @@ func (p *XdsProxy) handleUpstreamRequest(con *ProxyConnection) {
 }
 
 func (p *XdsProxy) handleUpstreamResponse(con *ProxyConnection) {
+	forwardEnvoyCh := make(chan *discovery.DiscoveryResponse, 1)
 	for {
 		select {
 		case resp := <-con.responsesChan:
@@ -506,7 +507,15 @@ func (p *XdsProxy) handleUpstreamResponse(con *ProxyConnection) {
 			case v3.ExtensionConfigurationType:
 				if features.WasmRemoteLoadConversion {
 					// If Wasm remote load conversion feature is enabled, rewrite and send.
-					go p.rewriteAndForward(con, resp)
+					go p.rewriteAndForward(con, resp, func(resp *discovery.DiscoveryResponse) {
+						// Forward the response using the thread of `handleUpstreamResponse`
+						// to prevent concurrent access to forwardToEnvoy
+						select {
+						case forwardEnvoyCh <- resp:
+							proxyLog.Debugf("wasm send response: %v", resp.TypeUrl)
+						case <-con.stopChan:
+						}
+					})
 				} else {
 					// Otherwise, forward ECDS resource update directly to Envoy.
 					forwardToEnvoy(con, resp)
@@ -518,13 +527,15 @@ func (p *XdsProxy) handleUpstreamResponse(con *ProxyConnection) {
 					forwardToEnvoy(con, resp)
 				}
 			}
+		case resp := <-forwardEnvoyCh:
+			forwardToEnvoy(con, resp)
 		case <-con.stopChan:
 			return
 		}
 	}
 }
 
-func (p *XdsProxy) rewriteAndForward(con *ProxyConnection, resp *discovery.DiscoveryResponse) {
+func (p *XdsProxy) rewriteAndForward(con *ProxyConnection, resp *discovery.DiscoveryResponse, forward func(resp *discovery.DiscoveryResponse)) {
 	sendNack := wasm.MaybeConvertWasmExtensionConfig(resp.Resources, p.wasmCache)
 	if sendNack {
 		proxyLog.Debugf("sending NACK for ECDS resources %+v", resp.Resources)
@@ -540,7 +551,7 @@ func (p *XdsProxy) rewriteAndForward(con *ProxyConnection, resp *discovery.Disco
 		return
 	}
 	proxyLog.Debugf("forward ECDS resources %+v", resp.Resources)
-	forwardToEnvoy(con, resp)
+	forward(resp)
 }
 
 func (p *XdsProxy) forwardToTap(resp *discovery.DiscoveryResponse) {