fix(eventlistener): remove goroutine from updateSnapshot

Rename updateSnapshotAsync to updateSnapshot and remove the
go func() wrapper. The event loop is already serial, and the
mutex on SnapshotManager now protects concurrent callers from
other paths (certificates, control plane sync).

Author: tricktron

gateway/gateway-controller/pkg/eventlistener/api_processor.go

@@ -43,21 +43,19 @@ func (l *EventListener) processAPIEvent(event eventhub.Event) {
 	}
 }
 
-func (l *EventListener) updateSnapshotAsync(entityID, correlationID, failureMessage string) {
+func (l *EventListener) updateSnapshot(entityID, correlationID, failureMessage string) {
 	if l.snapshotManager == nil {
 		return
 	}
 
-	go func() {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer cancel()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 
-		if err := l.snapshotManager.UpdateSnapshot(ctx, correlationID); err != nil {
-			l.logger.Error(failureMessage,
-				slog.String("entity_id", entityID),
-				slog.Any("error", err))
-		}
-	}()
+	if err := l.snapshotManager.UpdateSnapshot(ctx, correlationID); err != nil {
+		l.logger.Error(failureMessage,
+			slog.String("entity_id", entityID),
+			slog.Any("error", err))
+	}
 }
 
 // handleAPICreateOrUpdate handles API create or update events
@@ -109,7 +107,7 @@ func (l *EventListener) handleAPICreateOrUpdate(event eventhub.Event) {
 
 	// Update xDS snapshot for REST APIs only (WebSubApi and WebBrokerApi use Policy xDS)
 	if storedConfig.Kind != models.KindWebSubApi && storedConfig.Kind != models.KindWebBrokerApi {
-		l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after replica sync")
+		l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after replica sync")
 	}
 
 	// Update policies
@@ -188,7 +186,7 @@ func (l *EventListener) handleAPIDelete(event eventhub.Event) {
 
 	// Update xDS snapshot for REST APIs only (WebSubApi and WebBrokerApi use Policy xDS)
 	if existingConfig == nil || (existingConfig.Kind != models.KindWebSubApi && existingConfig.Kind != models.KindWebBrokerApi) {
-		l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after API deletion")
+		l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after API deletion")
 	}
 
 	// Remove runtime config for the deleted API

gateway/gateway-controller/pkg/eventlistener/llm_provider_processor.go

@@ -122,7 +122,7 @@ func (l *EventListener) handleLLMProviderCreateOrUpdate(event eventhub.Event) {
 			slog.Any("error", err))
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider replica sync")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider replica sync")
 	l.updatePoliciesForAPI(storedConfig, event.EventID)
 
 	l.logger.Info("Successfully processed LLM provider create/update event",
@@ -184,7 +184,7 @@ func (l *EventListener) handleLLMProxyCreateOrUpdate(event eventhub.Event) {
 		}
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy replica sync")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy replica sync")
 	l.updatePoliciesForAPI(storedConfig, event.EventID)
 
 	l.logger.Info("Successfully processed LLM proxy create/update event",
@@ -243,7 +243,7 @@ func (l *EventListener) handleLLMProviderDelete(event eventhub.Event) {
 		}
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider deletion")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider deletion")
 
 	if l.policyManager != nil && existingConfig != nil {
 		if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {
@@ -300,7 +300,7 @@ func (l *EventListener) handleLLMProxyDelete(event eventhub.Event) {
 		}
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy deletion")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy deletion")
 
 	if l.policyManager != nil && existingConfig != nil {
 		if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {

gateway/gateway-controller/pkg/eventlistener/mcp_processor.go

@@ -95,7 +95,7 @@ func (l *EventListener) handleMCPProxyCreateOrUpdate(event eventhub.Event) {
 		}
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy replica sync")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy replica sync")
 	l.updatePoliciesForAPI(storedConfig, event.EventID)
 
 	l.logger.Info("Successfully processed MCP proxy create/update event",
@@ -119,7 +119,7 @@ func (l *EventListener) handleMCPProxyDelete(event eventhub.Event) {
 		return
 	}
 
-	l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy deletion")
+	l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy deletion")
 
 	if l.policyManager != nil && existingConfig != nil {
 		if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {

gateway/gateway-controller/pkg/xds/snapshot_test.go

@@ -107,9 +107,9 @@ func TestConcurrentUpdateSnapshot(t *testing.T) {
 			close(aGotAll)
 			select {
 			case <-bDone:
-				// no mutex — B finished first, A will overwrite with stale data
+				// pre-fix path: B raced past A → stale overwrite will occur
 			case <-time.After(200 * time.Millisecond):
-				// mutex held — B is waiting for the lock, A continues
+				// post-fix path: mutex held, B is queued behind A
 			}
 		}