Fix concurrency races in CancellableFuture and Options setMethodOptions

core/src/main/java/feign/AsynchronousMethodHandler.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.TimeUnit;
+
 import java.util.function.BiConsumer;
 import java.util.stream.Stream;
 
@@ -114,18 +115,31 @@ private CompletableFuture<Object> executeAndDecode(
   }
 
   private static class CancellableFuture<T> extends CompletableFuture<T> {
-    private CompletableFuture<T> inner = null;
-
+    // volatile provides the same JMM happens-before guarantees as AtomicReference
+    // since we only ever read/write (never CAS), with less indirection.
+    private volatile CompletableFuture<T> inner;
+
+    /**
+     * Registers {@code value} as the active inner future and pipes its result into this future.
+     *
+     * <p>Side-effect: if this future has already been cancelled before {@code setInner} is called,
+     * the cancellation is immediately forwarded to {@code value} so that the in-flight async work
+     * is also cancelled rather than completing silently.
+     */
     public void setInner(CompletableFuture<T> value) {
       inner = value;
-      inner.whenComplete(pipeTo(this));
+      value.whenComplete(pipeTo(this));
+      if (isCancelled()) {
+        value.cancel(true);
+      }
     }
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
       final boolean result = super.cancel(mayInterruptIfRunning);
-      if (inner != null) {
-        inner.cancel(mayInterruptIfRunning);
+      CompletableFuture<T> current = inner;
+      if (current != null) {
+        current.cancel(mayInterruptIfRunning);
       }
       return result;
     }

core/src/main/java/feign/Request.java

@@ -26,7 +26,6 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
@@ -327,6 +326,15 @@ public static class Options {
     private final boolean followRedirects;
     private final Map<String, Map<String, Options>> threadToMethodOptions;
 
+    /**
+     * Returns the identifier used to bucket method-level options by calling context. Defaults to
+     * the current thread's identity. Subclasses may override this to provide a fixed identifier,
+     * which is useful in tests to force concurrent threads to contend on the same outer map key.
+     */
+    protected String threadIdentifier() {
+      return getThreadIdentifier();
+    }
+
     /**
      * Get an Options by methodName
      *
@@ -335,9 +343,12 @@ public static class Options {
      */
     @Experimental
     public Options getMethodOptions(String methodName) {
-      Map<String, Options> methodOptions =
-          threadToMethodOptions.getOrDefault(getThreadIdentifier(), new HashMap<>());
-      return methodOptions.getOrDefault(methodName, this);
+      Map<String, Options> methodOptions = threadToMethodOptions.get(threadIdentifier());
+      if (methodOptions == null) {
+        return this;
+      }
+      Options options = methodOptions.get(methodName);
+      return options != null ? options : this;
     }
 
     /**
@@ -348,11 +359,9 @@ public Options getMethodOptions(String methodName) {
      */
     @Experimental
     public void setMethodOptions(String methodName, Options options) {
-      String threadIdentifier = getThreadIdentifier();
-      Map<String, Request.Options> methodOptions =
-          threadToMethodOptions.getOrDefault(threadIdentifier, new HashMap<>());
-      threadToMethodOptions.put(threadIdentifier, methodOptions);
-      methodOptions.put(methodName, options);
+      threadToMethodOptions
+          .computeIfAbsent(threadIdentifier(), key -> new ConcurrentHashMap<>())
+          .put(methodName, options);
     }
 
     /**
@@ -517,10 +526,10 @@ public static class Body implements Serializable {
 
     private transient Charset encoding;
 
-    private byte[] data;
+    private final byte[] data;
 
     private Body() {
-      super();
+      this(null);
     }
 
     private Body(byte[] data) {

core/src/test/java/feign/CancellableFutureTest.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright © 2012 The Feign Authors (feign@commonhaus.dev)
+ *
+ * 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 feign;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.Test;
+
+class CancellableFutureTest {
+
+  interface Api {
+    @RequestLine("GET /")
+    CompletableFuture<String> get();
+  }
+
+  /**
+   * cancel() arrives BEFORE setInner() is called.
+   *
+   * <p>The AsyncClient returns immediately with a pending CompletableFuture so that api.get()
+   * returns the CancellableFuture to the caller without blocking. The caller then cancels it before
+   * the client future is completed. When the client future eventually completes, setInner() must
+   * detect isCancelled() and immediately forward cancellation to the newly registered inner future.
+   */
+  @Test
+  void cancelBeforeSetInnerRacesCorrectly() throws Exception {
+    // execute() returns this immediately — no blocking inside execute()
+    CompletableFuture<Response> clientFuture = new CompletableFuture<>();
+
+    AsyncClient<Void> client = (request, options, ctx) -> clientFuture;
+
+    Api api = AsyncFeign.<Void>builder().client(client).target(Api.class, "http://localhost:0");
+
+    // api.get() returns immediately because execute() returns immediately
+    CompletableFuture<String> result = api.get();
+
+    // Cancel BEFORE clientFuture resolves — inner is not yet set on CancellableFuture
+    result.cancel(true);
+
+    // Complete the client future now. This triggers the whenComplete → setInner() path.
+    // setInner() must see isCancelled() == true and cancel the newly registered inner future.
+    clientFuture.complete(
+        Response.builder()
+            .status(200)
+            .reason("OK")
+            .request(
+                Request.create(
+                    Request.HttpMethod.GET,
+                    "http://localhost:0",
+                    Collections.emptyMap(),
+                    Request.Body.empty(),
+                    null))
+            .build());
+
+    assertThat(result).isCancelled();
+  }
+
+  /**
+   * cancel() arrives AFTER setInner() has already been called (the retry path).
+   *
+   * <p>The first execute() fails immediately to trigger a retry. The retry execute() returns a
+   * pending CompletableFuture immediately (no blocking inside execute()) and signals a latch so
+   * the caller knows setInner() has been called. The caller then cancels — cancel() must read
+   * inner and propagate to the retry future.
+   */
+  @Test
+  void cancelAfterSetInnerRacesCorrectly() throws Exception {
+    AtomicInteger callCount = new AtomicInteger();
+    CountDownLatch retryStarted = new CountDownLatch(1);
+    // Holds the raw client future from the retry execute() call
+    CompletableFuture<Response>[] retryFutureHolder = new CompletableFuture[1];
+
+    AsyncClient<Void> client =
+        (request, options, ctx) -> {
+          int n = callCount.incrementAndGet();
+          if (n == 1) {
+            // First call: fail immediately to trigger the retryer
+            CompletableFuture<Response> failed = new CompletableFuture<>();
+            failed.completeExceptionally(new IOException("transient"));
+            return failed;
+          }
+          // Retry call: return a pending future immediately — execute() does NOT block.
+          // The latch is used only to tell the caller that setInner() has been called.
+          CompletableFuture<Response> retryFuture = new CompletableFuture<>();
+          retryFutureHolder[0] = retryFuture;
+          retryStarted.countDown();
+          return retryFuture;
+        };
+
+    Api api =
+        AsyncFeign.<Void>builder()
+            .client(client)
+            .retryer(new Retryer.Default(0, 0, 2))
+            .target(Api.class, "http://localhost:0");
+
+    CompletableFuture<String> result = api.get();
+
+    // Wait until the retry execute() returned and setInner() has been called
+    assertThat(retryStarted.await(2, TimeUnit.SECONDS)).isTrue();
+
+    // cancel() now arrives after setInner() — inner is already set to retryFuture
+    result.cancel(true);
+
+    assertThat(result).isCancelled();
+
+    // Verify the pipeTo guard: even if the raw retry client future eventually completes,
+    // it must NOT overwrite the cancellation on result. setInner() registered a whenComplete
+    // that calls pipeTo(result), which checks isDone() before completing — so result must
+    // remain cancelled after the raw client future resolves.
+    CompletableFuture<Response> retryFuture = retryFutureHolder[0];
+    assertThat(retryFuture).isNotNull();
+    retryFuture.cancel(false); // let the retry future give up
+    assertThat(result).isCancelled(); // must still be cancelled, not overwritten
+  }
+
+  /** Normal completion (no cancellation) must not be disrupted by the volatile field change. */
+  @Test
+  void normalCompletionIsNotAffected() throws Exception {
+    MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setBody("hello"));
+
+    Api api = AsyncFeign.<Void>builder().target(Api.class, server.url("/").toString());
+
+    assertThat(api.get().get(2, TimeUnit.SECONDS)).isEqualTo("hello");
+    server.shutdown();
+  }
+}
+

core/src/test/java/feign/OptionsTest.java

@@ -38,6 +38,22 @@ public ChildOptions(int connectTimeoutMillis, int readTimeoutMillis) {
     }
   }
 
+  /**
+   * Options subclass that overrides threadIdentifier() to return a fixed constant, forcing all
+   * threads to contend on the same outer key in threadToMethodOptions. This is the only way to
+   * exercise the pre-fix check-then-act race without changing thread identity.
+   */
+  static class SharedKeyOptions extends Request.Options {
+    public SharedKeyOptions(int connectTimeoutMillis, int readTimeoutMillis) {
+      super(connectTimeoutMillis, readTimeoutMillis);
+    }
+
+    @Override
+    protected String threadIdentifier() {
+      return "shared-key";
+    }
+  }
+
   interface OptionsInterface {
     @RequestLine("GET /")
     String get(Request.Options options);
@@ -135,4 +151,50 @@ void normalResponseWithMethodOptionsTest() throws Exception {
     thread.start();
     thread.join();
   }
+
+  /**
+   * Forces multiple threads to contend on the SAME outer key in threadToMethodOptions by using
+   * SharedKeyOptions, which returns a fixed "shared-key" from threadIdentifier().
+   *
+   * <p>Before the fix (getOrDefault + put), two threads racing with the same key could both
+   * observe the key absent, both create a new inner map, and one thread's put would overwrite the
+   * other's — silently losing entries. With computeIfAbsent + ConcurrentHashMap, creation is
+   * atomic and all entries must be present after all threads complete.
+   */
+  @Test
+  void concurrentSetMethodOptionsOnSameKeyDoesNotLoseEntries() throws Exception {
+    SharedKeyOptions options = new SharedKeyOptions(1000, 1000);
+    int threadCount = 20;
+    CountDownLatch start = new CountDownLatch(1);
+    CountDownLatch done = new CountDownLatch(threadCount);
+    AtomicReference<Throwable> error = new AtomicReference<>();
+
+    for (int i = 0; i < threadCount; i++) {
+      final String method = "method" + i;
+      new Thread(
+              () -> {
+                try {
+                  start.await();
+                  options.setMethodOptions(method, new Request.Options(1000, 2000));
+                } catch (Throwable t) {
+                  error.set(t);
+                } finally {
+                  done.countDown();
+                }
+              })
+          .start();
+    }
+
+    start.countDown();
+    assertThat(done.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // No exception must have been thrown
+    assertThat(error.get()).isNull();
+    // All 20 entries must be present — proves no lost updates due to the race
+    for (int i = 0; i < threadCount; i++) {
+      assertThat(options.getMethodOptions("method" + i))
+          .as("entry for method%d must not have been lost", i)
+          .isNotSameAs(options);
+    }
+  }
 }