From f273d155f85aa5be275590d90c8f9d90ed3e2cb2 Mon Sep 17 00:00:00 2001 From: bcleenders Date: Fri, 2 Oct 2026 14:43:48 -0400 Subject: [PATCH 1/2] xds: add tests for `DistributorWatcher` thread safety `DistributorWatcher.updateSpiffeTrustMap()` is not `synchronized`, unlike `updateCertificate()` and `updateTrustedRoots()` in the same class. This causes `ConcurrentModificationException` when a watcher is added or removed while `updateSpiffeTrustMap` iterates `downstreamWatchers`. Additionally, all update methods iterate `downstreamWatchers` directly. If a callback reentrantly calls `addWatcher`/`removeWatcher`, the `HashSet` is modified during iteration, producing: ``` java.util.ConcurrentModificationException at java.util.HashMap$HashIterator.nextNode(HashMap.java:1606) at java.util.HashMap$KeyIterator.next(HashMap.java:1629) at DistributorWatcher.updateSpiffeTrustMap(CertificateProvider.java:120) ``` Observed in production when a service creates many mTLS-enabled gRPC channels concurrently at startup (~60 channels across 7 threads). All channels share the same `file_watcher` cert provider instance. The burst of concurrent `addWatcher()` calls races with the provider's initial `updateSpiffeTrustMap()` call, triggering the CME. The exception propagates to `DynamicSslContextProvider.onError()`, preventing the TLS context from being created. Affected subchannels remain in CONNECTING until the deadline expires: ``` DEADLINE_EXCEEDED: CallOptions deadline exceeded after 4.998s. Name resolution delay 0.000000000 seconds. [closed=[], open=[[connecting_and_lb_delay=4999ms, was_still_waiting]]] ``` These tests demonstrate the race. They are expected to fail until the fix is applied. --- .../certprovider/CertificateProviderTest.java | 192 ++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java diff --git a/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java new file mode 100644 index 00000000000..8dcf30100d8 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/security/certprovider/CertificateProviderTest.java @@ -0,0 +1,192 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 io.grpc.xds.internal.security.certprovider; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.grpc.Status; +import io.grpc.xds.internal.security.certprovider.CertificateProvider.DistributorWatcher; +import io.grpc.xds.internal.security.certprovider.CertificateProvider.Watcher; +import java.security.PrivateKey; +import java.security.cert.X509Certificate; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests notification delivery when callbacks change certificate provider subscriptions. */ +@RunWith(JUnit4.class) +public class CertificateProviderTest { + private final DistributorWatcher distributor = new DistributorWatcher(); + private final Watcher watcher1 = mock(Watcher.class); + private final Watcher watcher2 = mock(Watcher.class); + private final Map> trustMap = + ImmutableMap.of("example.org", ImmutableList.of(mock(X509Certificate.class))); + + @Test + public void updateSpiffeTrustMap_doesNotOverlapCertificateReplay() throws Exception { + PrivateKey key = mock(PrivateKey.class); + List certChain = ImmutableList.of(mock(X509Certificate.class)); + distributor.updateCertificate(key, certChain); + CountDownLatch replayStarted = new CountDownLatch(1); + CountDownLatch finishReplay = new CountDownLatch(1); + CountDownLatch updateStarted = new CountDownLatch(1); + AtomicBoolean replayingCertificate = new AtomicBoolean(); + AtomicBoolean overlappingNotifications = new AtomicBoolean(); + doAnswer(invocation -> { + replayingCertificate.set(true); + replayStarted.countDown(); + assertThat(finishReplay.await(10, TimeUnit.SECONDS)).isTrue(); + replayingCertificate.set(false); + return null; + }).when(watcher1).updateCertificate(key, certChain); + doAnswer(invocation -> { + overlappingNotifications.compareAndSet(false, replayingCertificate.get()); + return null; + }).when(watcher1).updateSpiffeTrustMap(trustMap); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future registration = executor.submit(() -> distributor.addWatcher(watcher1)); + assertThat(replayStarted.await(10, TimeUnit.SECONDS)).isTrue(); + Future update = executor.submit(() -> { + updateStarted.countDown(); + distributor.updateSpiffeTrustMap(trustMap); + }); + assertThat(updateStarted.await(10, TimeUnit.SECONDS)).isTrue(); + try { + // A serialized update waits for registration to finish replaying the certificate. + update.get(1, TimeUnit.SECONDS); + } catch (TimeoutException expected) { + // Release registration below so both tasks can finish before checking delivery. + } finally { + finishReplay.countDown(); + } + registration.get(10, TimeUnit.SECONDS); + update.get(10, TimeUnit.SECONDS); + + assertThat(overlappingNotifications.get()).isFalse(); + verify(watcher1).updateSpiffeTrustMap(trustMap); + } finally { + finishReplay.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + public void updateCertificate_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + PrivateKey key = mock(PrivateKey.class); + List certChain = ImmutableList.of(mock(X509Certificate.class)); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateCertificate(key, certChain); + + distributor.updateCertificate(key, certChain); + + verify(watcher1).updateCertificate(key, certChain); + verify(watcher2).updateCertificate(key, certChain); + } + + @Test + public void updateTrustedRoots_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + List trustedRoots = ImmutableList.of(mock(X509Certificate.class)); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateTrustedRoots(trustedRoots); + + distributor.updateTrustedRoots(trustedRoots); + + verify(watcher1).updateTrustedRoots(trustedRoots); + verify(watcher2).updateTrustedRoots(trustedRoots); + } + + @Test + public void updateSpiffeTrustMap_watcherAddedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher addedWatcher = mock(Watcher.class); + // Select the first callback without depending on HashSet's iteration order. At least one + // existing watcher still needs its initial trust map when this callback changes the set. + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.addWatcher(addedWatcher); + return null; + }).when(firstWatcher).updateSpiffeTrustMap(trustMap); + + distributor.updateSpiffeTrustMap(trustMap); + + verify(watcher1).updateSpiffeTrustMap(trustMap); + verify(watcher2).updateSpiffeTrustMap(trustMap); + verify(addedWatcher).updateSpiffeTrustMap(trustMap); + } + + @Test + public void updateSpiffeTrustMap_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).updateSpiffeTrustMap(trustMap); + + distributor.updateSpiffeTrustMap(trustMap); + + verify(watcher1).updateSpiffeTrustMap(trustMap); + verify(watcher2).updateSpiffeTrustMap(trustMap); + assertThat(distributor.getDownstreamWatchers()).doesNotContain(firstWatcher); + } + + @Test + public void onError_watcherRemovedDuringNotification() { + distributor.addWatcher(watcher1); + distributor.addWatcher(watcher2); + Watcher firstWatcher = distributor.getDownstreamWatchers().iterator().next(); + doAnswer(invocation -> { + distributor.removeWatcher(firstWatcher); + return null; + }).when(firstWatcher).onError(Status.UNAVAILABLE); + + distributor.onError(Status.UNAVAILABLE); + + verify(watcher1).onError(Status.UNAVAILABLE); + verify(watcher2).onError(Status.UNAVAILABLE); + assertThat(distributor.getDownstreamWatchers()).doesNotContain(firstWatcher); + } +} From 0713c232b6c6e0351d78f3d6b7535b6fc84e27e3 Mon Sep 17 00:00:00 2001 From: bcleenders Date: Fri, 2 Oct 2026 14:47:22 -0400 Subject: [PATCH 2/2] xds: fix `DistributorWatcher` thread safety Add `synchronized` to `updateSpiffeTrustMap()`, matching `updateCertificate()` and `updateTrustedRoots()`. Iterate a snapshot (`ImmutableSet.copyOf`) of `downstreamWatchers` in all notification methods, instead of iterating the live `HashSet`. This prevents `ConcurrentModificationException` when a callback calls `addWatcher()`/`removeWatcher()`. Mark other functions (`getLastIdentityCert()`, `close()` & `clearValues()`) which read/write state without holding the monitor as synchronized. --- .../certprovider/CertificateProvider.java | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java b/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java index 009bb7bf566..7e31516b666 100644 --- a/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java +++ b/xds/src/main/java/io/grpc/xds/internal/security/certprovider/CertificateProvider.java @@ -19,11 +19,11 @@ import static com.google.common.base.Preconditions.checkNotNull; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableSet; import io.grpc.Status; import io.grpc.xds.internal.security.Closeable; import java.security.PrivateKey; import java.security.cert.X509Certificate; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -78,8 +78,9 @@ synchronized void removeWatcher(Watcher watcher) { downstreamWatchers.remove(watcher); } - @VisibleForTesting public Set getDownstreamWatchers() { - return Collections.unmodifiableSet(downstreamWatchers); + @VisibleForTesting public synchronized Set getDownstreamWatchers() { + // Callbacks may register or remove watchers reentrantly, even while we hold the lock. + return ImmutableSet.copyOf(downstreamWatchers); } private void sendLastCertificateUpdate(Watcher watcher) { @@ -100,7 +101,7 @@ public synchronized void updateCertificate(PrivateKey key, List checkNotNull(certChain, "certChain"); privateKey = key; this.certChain = certChain; - for (Watcher watcher : downstreamWatchers) { + for (Watcher watcher : getDownstreamWatchers()) { sendLastCertificateUpdate(watcher); } } @@ -109,39 +110,40 @@ public synchronized void updateCertificate(PrivateKey key, List public synchronized void updateTrustedRoots(List trustedRoots) { checkNotNull(trustedRoots, "trustedRoots"); this.trustedRoots = trustedRoots; - for (Watcher watcher : downstreamWatchers) { + for (Watcher watcher : getDownstreamWatchers()) { sendLastTrustedRootsUpdate(watcher); } } @Override - public void updateSpiffeTrustMap(Map> spiffeTrustMap) { + public synchronized void updateSpiffeTrustMap( + Map> spiffeTrustMap) { this.spiffeTrustMap = spiffeTrustMap; - for (Watcher watcher : downstreamWatchers) { + for (Watcher watcher : getDownstreamWatchers()) { sendLastSpiffeTrustMapUpdate(watcher); } } @Override public synchronized void onError(Status errorStatus) { - for (Watcher watcher : downstreamWatchers) { + for (Watcher watcher : getDownstreamWatchers()) { watcher.onError(errorStatus); } } - X509Certificate getLastIdentityCert() { + synchronized X509Certificate getLastIdentityCert() { if (certChain != null && !certChain.isEmpty()) { return certChain.get(0); } return null; } - void close() { + synchronized void close() { downstreamWatchers.clear(); clearValues(); } - void clearValues() { + synchronized void clearValues() { privateKey = null; certChain = null; trustedRoots = null;