From c67cd631924a3d9540b6a62de7f3669e8caaf6c5 Mon Sep 17 00:00:00 2001 From: Thibault Gagnaux Date: Sat, 18 Jul 2026 10:38:39 +0200 Subject: [PATCH 1/4] test(xds): add failing test for concurrent UpdateSnapshot race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a deterministic test that reproduces the TOCTOU race in UpdateSnapshot. A test hook forces goroutine A to read a stale store while goroutine B writes the correct snapshot — then A overwrites it with a higher version number. The test is RED: api-two is missing from the final snapshot. --- .../gateway-controller/pkg/xds/snapshot.go | 4 + .../pkg/xds/snapshot_test.go | 175 ++++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 gateway/gateway-controller/pkg/xds/snapshot_test.go diff --git a/gateway/gateway-controller/pkg/xds/snapshot.go b/gateway/gateway-controller/pkg/xds/snapshot.go index 8d474d06c8..ea1a564af4 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot.go +++ b/gateway/gateway-controller/pkg/xds/snapshot.go @@ -69,6 +69,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 @@ -113,6 +114,9 @@ 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) 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 0000000000..f8d176cb45 --- /dev/null +++ b/gateway/gateway-controller/pkg/xds/snapshot_test.go @@ -0,0 +1,175 @@ +/* + * 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" + + 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: A blocks until B finishes, then re-reads → 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 calls UpdateSnapshot, blocks after GetAll([api-one]) + 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) + <-bDone + } + + 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 := <-errB; err != nil { + t.Fatalf("goroutine B UpdateSnapshot: %v", err) + } + if err := <-errA; err != nil { + t.Fatalf("goroutine A 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) + } + } +} From a61706b4ff10eeaedcb507954c884c39cf2a8b9b Mon Sep 17 00:00:00 2001 From: Thibault Gagnaux Date: Sat, 18 Jul 2026 10:43:25 +0200 Subject: [PATCH 2/4] fix(xds): add mutex to SnapshotManager.UpdateSnapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Protect the GetAll → Translate → IncrementVersion → SetSnapshot sequence with sync.Mutex, matching the pattern used by the other four snapshot managers (subscription, apikey, lazyresource, policy). --- .../gateway-controller/pkg/xds/snapshot.go | 5 +++++ .../pkg/xds/snapshot_test.go | 19 ++++++++++++++----- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/gateway/gateway-controller/pkg/xds/snapshot.go b/gateway/gateway-controller/pkg/xds/snapshot.go index ea1a564af4..d97c6e2942 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 @@ -101,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 != "" { diff --git a/gateway/gateway-controller/pkg/xds/snapshot_test.go b/gateway/gateway-controller/pkg/xds/snapshot_test.go index f8d176cb45..6089e10fb1 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot_test.go +++ b/gateway/gateway-controller/pkg/xds/snapshot_test.go @@ -23,6 +23,7 @@ import ( "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" @@ -80,7 +81,10 @@ func makeRestAPI(uuid, name, ctx string) *models.StoredConfig { // Step 4: A: SetSnapshot(v3, [api-one]) ✗ api-two lost // // Without mutex: A wins (higher version, stale data) → FAIL -// With mutex: A blocks until B finishes, then re-reads → PASS +// With mutex: B can't run while A holds the lock. A finishes first +// +// (v1, [api-one]), then B runs (v2, [api-one, api-two]). +// B wins. → PASS func TestConcurrentUpdateSnapshot(t *testing.T) { t.Run("stale GetAll cannot overwrite a newer complete snapshot", func(t *testing.T) { metrics.Init() @@ -103,7 +107,12 @@ func TestConcurrentUpdateSnapshot(t *testing.T) { return } close(aGotAll) - <-bDone + select { + case <-bDone: + // B completed while A was paused — no mutex, bug is present + case <-time.After(200 * time.Millisecond): + // B couldn't run (blocked on mutex) — A continues + } } errA := make(chan error, 1) @@ -126,12 +135,12 @@ func TestConcurrentUpdateSnapshot(t *testing.T) { }() // Step 4: A resumes with stale [api-one], overwrites B's snapshot - if err := <-errB; err != nil { - t.Fatalf("goroutine B UpdateSnapshot: %v", err) - } 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"}) }) From 641d92f910b15614693d25bd447de47eaa7ae4d1 Mon Sep 17 00:00:00 2001 From: Thibault Gagnaux Date: Sat, 18 Jul 2026 11:12:55 +0200 Subject: [PATCH 3/4] refactor(xds): remove dead code, clean up test comments --- gateway/gateway-controller/pkg/xds/snapshot.go | 3 --- gateway/gateway-controller/pkg/xds/snapshot_test.go | 12 +++++------- 2 files changed, 5 insertions(+), 10 deletions(-) diff --git a/gateway/gateway-controller/pkg/xds/snapshot.go b/gateway/gateway-controller/pkg/xds/snapshot.go index d97c6e2942..bfd14c872d 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot.go +++ b/gateway/gateway-controller/pkg/xds/snapshot.go @@ -123,9 +123,6 @@ func (sm *SnapshotManager) UpdateSnapshot(ctx context.Context, correlationID str 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 index 6089e10fb1..944f45c984 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot_test.go +++ b/gateway/gateway-controller/pkg/xds/snapshot_test.go @@ -81,10 +81,8 @@ func makeRestAPI(uuid, name, ctx string) *models.StoredConfig { // 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 -// -// (v1, [api-one]), then B runs (v2, [api-one, api-two]). -// B wins. → PASS +// 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() @@ -96,7 +94,7 @@ func TestConcurrentUpdateSnapshot(t *testing.T) { sm := NewSnapshotManager(store, createTestLogger(), testRouterConfig(), nil, testConfig()) - // Step 1: A calls UpdateSnapshot, blocks after GetAll([api-one]) + // Step 1: A reads store, blocks in hook aGotAll := make(chan struct{}) bDone := make(chan struct{}) @@ -109,9 +107,9 @@ func TestConcurrentUpdateSnapshot(t *testing.T) { close(aGotAll) select { case <-bDone: - // B completed while A was paused — no mutex, bug is present + // no mutex — B finished first, A will overwrite with stale data case <-time.After(200 * time.Millisecond): - // B couldn't run (blocked on mutex) — A continues + // mutex held — B is waiting for the lock, A continues } } From 50f94bfd9cf48c0b2b058b9982a8902ae90333e7 Mon Sep 17 00:00:00 2001 From: Thibault Gagnaux Date: Sat, 18 Jul 2026 11:16:25 +0200 Subject: [PATCH 4/4] 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). --- .../pkg/eventlistener/api_processor.go | 22 +++++++++---------- .../eventlistener/llm_provider_processor.go | 8 +++---- .../pkg/eventlistener/mcp_processor.go | 4 ++-- .../pkg/xds/snapshot_test.go | 4 ++-- 4 files changed, 18 insertions(+), 20 deletions(-) diff --git a/gateway/gateway-controller/pkg/eventlistener/api_processor.go b/gateway/gateway-controller/pkg/eventlistener/api_processor.go index 2905caad63..5d97a766be 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 3383b7ccc3..2f69b68f2e 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 53a45f12ca..983224af34 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_test.go b/gateway/gateway-controller/pkg/xds/snapshot_test.go index 944f45c984..743a3e9ab2 100644 --- a/gateway/gateway-controller/pkg/xds/snapshot_test.go +++ b/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 } }