diff --git a/gateway/gateway-controller/pkg/eventlistener/api_processor.go b/gateway/gateway-controller/pkg/eventlistener/api_processor.go index 2905caad6..5d97a766b 100644 --- a/gateway/gateway-controller/pkg/eventlistener/api_processor.go +++ b/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 diff --git a/gateway/gateway-controller/pkg/eventlistener/llm_provider_processor.go b/gateway/gateway-controller/pkg/eventlistener/llm_provider_processor.go index 3383b7ccc..2f69b68f2 100644 --- a/gateway/gateway-controller/pkg/eventlistener/llm_provider_processor.go +++ b/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 { diff --git a/gateway/gateway-controller/pkg/eventlistener/mcp_processor.go b/gateway/gateway-controller/pkg/eventlistener/mcp_processor.go index 53a45f12c..983224af3 100644 --- a/gateway/gateway-controller/pkg/eventlistener/mcp_processor.go +++ b/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 { diff --git a/gateway/gateway-controller/pkg/xds/snapshot.go b/gateway/gateway-controller/pkg/xds/snapshot.go index 8d474d06c..bfd14c872 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot.go +++ b/gateway/gateway-controller/pkg/xds/snapshot.go @@ -22,6 +22,7 @@ import ( "context" "fmt" "log/slog" + "sync" "time" "github.com/envoyproxy/go-control-plane/pkg/cache/types" @@ -62,6 +63,7 @@ type StatusUpdateCallback func(configID string, success bool, correlationID stri // SnapshotManager manages xDS snapshots for Envoy type SnapshotManager struct { + mu sync.Mutex cache cache.SnapshotCache translator *Translator store *storage.ConfigStore @@ -69,6 +71,7 @@ type SnapshotManager struct { nodeID string // Node ID for Envoy (default: "router-node") statusCallback StatusUpdateCallback sdsSecretManager *SDSSecretManager + afterGetAll func() // nil in production; test hook for deterministic race testing } // NewSnapshotManager creates a new snapshot manager @@ -100,6 +103,9 @@ func (sm *SnapshotManager) SetStatusCallback(callback StatusUpdateCallback) { // UpdateSnapshot generates a new xDS snapshot from all configurations and updates the cache // The correlationID parameter is optional and used for request tracing in logs func (sm *SnapshotManager) UpdateSnapshot(ctx context.Context, correlationID string) error { + sm.mu.Lock() + defer sm.mu.Unlock() + startTime := time.Now() trigger := "manual" if correlationID != "" { @@ -113,10 +119,10 @@ func (sm *SnapshotManager) UpdateSnapshot(ctx context.Context, correlationID str } // Get all configurations from in-memory store configs := sm.store.GetAll() + if sm.afterGetAll != nil { + sm.afterGetAll() + } - // Translate configurations to Envoy resources if this is not event gw - //resources, err := sm.translator.TranslateConfigs(configs, correlationID) - // If event gw, resources, err := sm.translator.TranslateConfigs(configs, correlationID) if err != nil { log.Error("Failed to translate configurations", slog.Any("error", err)) diff --git a/gateway/gateway-controller/pkg/xds/snapshot_test.go b/gateway/gateway-controller/pkg/xds/snapshot_test.go new file mode 100644 index 000000000..743a3e9ab --- /dev/null +++ b/gateway/gateway-controller/pkg/xds/snapshot_test.go @@ -0,0 +1,182 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com). + * + * WSO2 LLC. licenses this file to you under the Apache License, + * Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package xds + +import ( + "context" + "strings" + "sync/atomic" + "testing" + "time" + + route "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" + "github.com/envoyproxy/go-control-plane/pkg/resource/v3" + api "github.com/wso2/api-platform/gateway/gateway-controller/pkg/api/management" + "github.com/wso2/api-platform/gateway/gateway-controller/pkg/metrics" + "github.com/wso2/api-platform/gateway/gateway-controller/pkg/models" + "github.com/wso2/api-platform/gateway/gateway-controller/pkg/storage" +) + +func makeRestAPI(uuid, name, ctx string) *models.StoredConfig { + cfg := api.RestAPI{ + Kind: api.RestAPIKindRestApi, + Metadata: api.Metadata{Name: name}, + Spec: api.APIConfigData{ + DisplayName: name, + Version: "v1.0", + Context: ctx, + Upstream: struct { + Main api.Upstream `json:"main" yaml:"main"` + Sandbox *api.Upstream `json:"sandbox,omitempty" yaml:"sandbox,omitempty"` + }{ + Main: api.Upstream{Url: api.Ptr("http://backend:8080")}, + }, + Operations: []api.Operation{ + {Method: api.Ptr(api.OperationMethodGET), Path: api.Ptr("/resource")}, + }, + }, + } + return &models.StoredConfig{ + UUID: uuid, + Kind: models.KindRestApi, + Handle: name, + DisplayName: name, + Version: "v1.0", + DesiredState: models.StateDeployed, + Configuration: cfg, + SourceConfiguration: cfg, + } +} + +// TestConcurrentUpdateSnapshot reproduces a race where two goroutines call +// UpdateSnapshot and the final snapshot only contains api-one instead of +// both api-one and api-two. The slower goroutine reads the store before +// api-two is added, but writes last with a higher version, overwriting +// the correct snapshot. +// +// A test hook forces this interleaving: +// +// Step 1: A: GetAll() → [api-one] +// ↓ hook blocks A +// Step 2: Main: store.Add(api-two) +// Step 3: B: GetAll() → [api-one, api-two] +// SetSnapshot(v2, both) ✓ +// ↓ hook releases A +// Step 4: A: SetSnapshot(v3, [api-one]) ✗ api-two lost +// +// Without mutex: A wins (higher version, stale data) → FAIL +// With mutex: B can't run while A holds the lock. A finishes first, +// then B reads the full store and writes the final snapshot → PASS +func TestConcurrentUpdateSnapshot(t *testing.T) { + t.Run("stale GetAll cannot overwrite a newer complete snapshot", func(t *testing.T) { + metrics.Init() + store := storage.NewConfigStore() + + if err := store.Add(makeRestAPI("uuid-api-1", "api-one", "/api-one")); err != nil { + t.Fatalf("Add api-one: %v", err) + } + + sm := NewSnapshotManager(store, createTestLogger(), testRouterConfig(), nil, testConfig()) + + // Step 1: A reads store, blocks in hook + aGotAll := make(chan struct{}) + bDone := make(chan struct{}) + + // Only block the first caller (A); let B pass through. + var hooked atomic.Bool + sm.afterGetAll = func() { + if !hooked.CompareAndSwap(false, true) { + return + } + close(aGotAll) + select { + case <-bDone: + // pre-fix path: B raced past A → stale overwrite will occur + case <-time.After(200 * time.Millisecond): + // post-fix path: mutex held, B is queued behind A + } + } + + errA := make(chan error, 1) + go func() { + errA <- sm.UpdateSnapshot(context.Background(), "corr-A") + }() + <-aGotAll + + // Step 2: add api-two behind A's back + if err := store.Add(makeRestAPI("uuid-api-2", "api-two", "/api-two")); err != nil { + t.Fatalf("Add api-two: %v", err) + } + + // Step 3: B sees [api-one, api-two], completes, releases A + errB := make(chan error, 1) + go func() { + err := sm.UpdateSnapshot(context.Background(), "corr-B") + close(bDone) + errB <- err + }() + + // Step 4: A resumes with stale [api-one], overwrites B's snapshot + if err := <-errA; err != nil { + t.Fatalf("goroutine A UpdateSnapshot: %v", err) + } + if err := <-errB; err != nil { + t.Fatalf("goroutine B UpdateSnapshot: %v", err) + } + + assertSnapshotContainsAPIs(t, sm, []string{"/api-one", "/api-two"}) + }) +} + +func assertSnapshotContainsAPIs(t *testing.T, sm *SnapshotManager, expectedContexts []string) { + t.Helper() + + snap, err := sm.GetCache().GetSnapshot("router-node") + if err != nil { + t.Fatalf("GetSnapshot: %v", err) + } + + var routePaths []string + for _, res := range snap.GetResources(resource.RouteType) { + routeCfg, ok := res.(*route.RouteConfiguration) + if !ok { + continue + } + for _, vh := range routeCfg.GetVirtualHosts() { + for _, r := range vh.GetRoutes() { + if regex, ok := r.GetMatch().GetPathSpecifier().(*route.RouteMatch_SafeRegex); ok { + routePaths = append(routePaths, regex.SafeRegex.GetRegex()) + } + } + } + } + + for _, ctx := range expectedContexts { + found := false + for _, p := range routePaths { + if strings.Contains(p, ctx) { + found = true + break + } + } + if !found { + t.Errorf("snapshot after concurrent UpdateSnapshot is missing %s; got routes: %v", ctx, routePaths) + } + } +}