Fix flaky EventListenerTest_Relay.cancelAsyncCall

okhttp-testing-support/src/main/kotlin/okhttp3/EventListenerRelay.kt

@@ -15,6 +15,10 @@
  */
 package okhttp3
 
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
+
 /**
  * A special [EventListener] for testing the mechanics of event listeners.
  *
@@ -23,11 +27,21 @@ package okhttp3
  *
  * By forcing the list of listeners to change after every event, we can detect if buggy code caches
  * a stale [EventListener] in a field or local variable.
+ *
+ * If a second event arrives while this instance is still handing off to its successor (for example
+ * [EventListener.canceled] racing a connection event), it is queued and flushed to the successor
+ * so it is not dropped.
  */
-class EventListenerRelay(
+class EventListenerRelay private constructor(
   val call: Call,
   val eventRecorder: EventRecorder,
+  private val state: State,
 ) {
+  constructor(
+    call: Call,
+    eventRecorder: EventRecorder,
+  ) : this(call, eventRecorder, State())
+
   private val eventListenerAdapter =
     EventListenerAdapter()
       .apply {
@@ -37,13 +51,34 @@ class EventListenerRelay(
   val eventListener: EventListener
     get() = eventListenerAdapter
 
-  var eventCount = 0
+  private val accepted = AtomicBoolean()
+
+  init {
+    state.tail.compareAndSet(null, this)
+  }
 
   private fun onEvent(callEvent: CallEvent) {
-    if (eventCount++ == 0) {
-      eventRecorder.logEvent(callEvent)
-      val next = EventListenerRelay(call, eventRecorder)
-      call.addEventListener(next.eventListener)
+    // Older relays remain in the aggregate as no-ops so cached listeners stop receiving events.
+    if (this !== state.tail.get()) return
+
+    if (!accepted.compareAndSet(false, true)) {
+      state.pending.offer(callEvent)
+      return
     }
+
+    val next = EventListenerRelay(call, eventRecorder, state)
+    call.addEventListener(next.eventListener)
+    eventRecorder.logEvent(callEvent)
+    state.tail.set(next)
+
+    while (true) {
+      val pending = state.pending.poll() ?: break
+      next.onEvent(pending)
+    }
+  }
+
+  private class State {
+    val tail = AtomicReference<EventListenerRelay?>(null)
+    val pending = ConcurrentLinkedQueue<CallEvent>()
   }
 }

okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt

@@ -125,12 +125,14 @@ open class EventRecorder(
       assertThat(Thread.holdsLock(lock), lock.toString()).isFalse()
     }
 
-    if (enforceOrder) {
-      checkForStartEvent(e)
-    }
+    synchronized(this) {
+      if (enforceOrder) {
+        checkForStartEvent(e)
+      }
 
-    eventsForMatching.offer(e)
-    eventSequence.offer(e)
+      eventsForMatching.offer(e)
+      eventSequence.offer(e)
+    }
   }
 
   private fun checkForStartEvent(e: CallEvent) {

okhttp-testing-support/src/test/kotlin/okhttp3/EventListenerRelayTest.kt

@@ -0,0 +1,149 @@
+/*
+ * Copyright (C) 2026 Square, Inc.
+ *
+ * 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 okhttp3
+
+import assertk.assertThat
+import assertk.assertions.contains
+import assertk.assertions.containsExactly
+import assertk.assertions.hasSize
+import assertk.assertions.isTrue
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater
+import kotlin.reflect.KClass
+import okhttp3.CallEvent.CallStart
+import okhttp3.CallEvent.Canceled
+import okio.Timeout
+import org.junit.jupiter.api.Test
+
+/** Regression coverage for #9372 (cancel during EventListenerRelay handoff). */
+class EventListenerRelayTest {
+  @Test
+  fun sequentialEventsAreRecordedOnceEach() {
+    val call = RecordingCall()
+    val eventRecorder = EventRecorder()
+    call.addEventListener(EventListenerRelay(call, eventRecorder).eventListener)
+
+    call.eventListener.callStart(call)
+    call.eventListener.canceled(call)
+
+    assertThat(eventRecorder.recordedEventTypes()).containsExactly(
+      CallStart::class,
+      Canceled::class,
+    )
+  }
+
+  /**
+   * Cancel while the active relay is blocked installing its successor. The pre-fix one-shot relay
+   * dropped [Canceled] in this window; the recorder must still see both events.
+   */
+  @Test
+  fun cancelDuringSuccessorHandoffIsRecorded() {
+    val startedSuccessorAdd = CountDownLatch(1)
+    val finishSuccessorAdd = CountDownLatch(1)
+    val addCount = AtomicInteger()
+
+    val call =
+      object : RecordingCall() {
+        override fun addEventListener(eventListener: EventListener) {
+          // Block successor installs only (not the initial relay registration).
+          if (addCount.getAndIncrement() > 0) {
+            startedSuccessorAdd.countDown()
+            assertThat(finishSuccessorAdd.await(5, TimeUnit.SECONDS)).isTrue()
+          }
+          super.addEventListener(eventListener)
+        }
+      }
+
+    val eventRecorder = EventRecorder()
+    call.addEventListener(EventListenerRelay(call, eventRecorder).eventListener)
+
+    val errors = AtomicReference<Throwable>()
+    val callStartDone = CountDownLatch(1)
+    Thread {
+      try {
+        call.eventListener.callStart(call)
+      } catch (t: Throwable) {
+        errors.compareAndSet(null, t)
+      } finally {
+        callStartDone.countDown()
+      }
+    }.start()
+
+    assertThat(startedSuccessorAdd.await(5, TimeUnit.SECONDS)).isTrue()
+    call.eventListener.canceled(call)
+    finishSuccessorAdd.countDown()
+    assertThat(callStartDone.await(5, TimeUnit.SECONDS)).isTrue()
+    errors.get()?.let { throw it }
+
+    val types = eventRecorder.recordedEventTypes()
+    assertThat(types).contains(CallStart::class)
+    assertThat(types).contains(Canceled::class)
+    assertThat(types).hasSize(2)
+  }
+
+  open class RecordingCall : Call {
+    @Volatile
+    var eventListener: EventListener = EventListener.NONE
+
+    override fun request(): Request = error("unexpected")
+
+    override fun execute(): Response = error("unexpected")
+
+    override fun enqueue(responseCallback: Callback): Unit = error("unexpected")
+
+    override fun cancel(): Unit = error("unexpected")
+
+    override fun isExecuted(): Boolean = error("unexpected")
+
+    override fun isCanceled(): Boolean = error("unexpected")
+
+    override fun timeout(): Timeout = error("unexpected")
+
+    override fun addEventListener(eventListener: EventListener) {
+      do {
+        val previous = this.eventListener
+      } while (!eventListenerUpdater.compareAndSet(this, previous, previous + eventListener))
+    }
+
+    override fun <T : Any> tag(type: KClass<T>): T? = error("unexpected")
+
+    override fun <T> tag(type: Class<out T>): T? = error("unexpected")
+
+    override fun <T : Any> tag(
+      type: KClass<T>,
+      computeIfAbsent: () -> T,
+    ): T = error("unexpected")
+
+    override fun <T : Any> tag(
+      type: Class<T>,
+      computeIfAbsent: () -> T,
+    ): T = error("unexpected")
+
+    override fun clone(): Call = error("unexpected")
+
+    companion object {
+      val eventListenerUpdater: AtomicReferenceFieldUpdater<RecordingCall, EventListener> =
+        AtomicReferenceFieldUpdater.newUpdater(
+          RecordingCall::class.java,
+          EventListener::class.java,
+          "eventListener",
+        )
+    }
+  }
+}

okhttp/src/jvmTest/kotlin/okhttp3/EventListenerTest.kt

@@ -30,6 +30,7 @@ import assertk.assertions.isNotEmpty
 import assertk.assertions.isNotNull
 import assertk.assertions.isNull
 import assertk.assertions.isSameInstanceAs
+import assertk.assertions.isTrue
 import assertk.assertions.prop
 import java.io.File
 import java.io.IOException
@@ -419,30 +420,51 @@ class EventListenerTest(
         .body("abc")
         .build(),
     )
+
+    // Hold the call in an application interceptor so cancel runs after CallStart handoff completes.
+    // Otherwise cancel races connection events and can miss Canceled under EventListenerRelay (#9372).
+    val inFlight = CountDownLatch(1)
+    val release = CountDownLatch(1)
+    client =
+      client
+        .newBuilder()
+        .addInterceptor { chain ->
+          inFlight.countDown()
+          assertThat(release.await(5, TimeUnit.SECONDS)).isTrue()
+          chain.proceed(chain.request())
+        }.build()
     val call =
       client.newCallWithListener(
         Request
           .Builder()
           .url(server.url("/"))
           .build(),
       )
+    val done = CountDownLatch(1)
     call.enqueue(
       object : Callback {
         override fun onFailure(
           call: Call,
           e: IOException,
         ) {
+          done.countDown()
         }
 
         override fun onResponse(
           call: Call,
           response: Response,
         ) {
           response.close()
+          done.countDown()
         }
       },
     )
+
+    assertThat(inFlight.await(5, TimeUnit.SECONDS)).isTrue()
     call.cancel()
+    release.countDown()
+
+    assertThat(done.await(5, TimeUnit.SECONDS)).isTrue()
     assertThat(eventRecorder.recordedEventTypes()).contains(Canceled::class)
   }