From a524e64b03ddecd48033b6a4b690ebc980e91512 Mon Sep 17 00:00:00 2001 From: Kamil Krzywanski Date: Sun, 19 Jul 2026 17:48:43 +0200 Subject: [PATCH] Fix flaky EventListenerRelay cancel under concurrent handoff Cancel racing connection events could hit a spent one-shot relay and never record Canceled (especially under Loom). Queue concurrent events on the active tail until the successor is installed, serialize EventRecorder start/end matching, and hold cancelAsyncCall in an application interceptor until after CallStart. Closes #9372 --- .../main/kotlin/okhttp3/EventListenerRelay.kt | 47 +++++- .../src/main/kotlin/okhttp3/EventRecorder.kt | 12 +- .../kotlin/okhttp3/EventListenerRelayTest.kt | 149 ++++++++++++++++++ .../kotlin/okhttp3/EventListenerTest.kt | 22 +++ 4 files changed, 219 insertions(+), 11 deletions(-) create mode 100644 okhttp-testing-support/src/test/kotlin/okhttp3/EventListenerRelayTest.kt diff --git a/okhttp-testing-support/src/main/kotlin/okhttp3/EventListenerRelay.kt b/okhttp-testing-support/src/main/kotlin/okhttp3/EventListenerRelay.kt index a66de098b53a..6d37e1d8a2eb 100644 --- a/okhttp-testing-support/src/main/kotlin/okhttp3/EventListenerRelay.kt +++ b/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(null) + val pending = ConcurrentLinkedQueue() } } diff --git a/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt b/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt index 85a23acf08b0..cd8da9fbd7ea 100644 --- a/okhttp-testing-support/src/main/kotlin/okhttp3/EventRecorder.kt +++ b/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) { diff --git a/okhttp-testing-support/src/test/kotlin/okhttp3/EventListenerRelayTest.kt b/okhttp-testing-support/src/test/kotlin/okhttp3/EventListenerRelayTest.kt new file mode 100644 index 000000000000..fdb8a408b3da --- /dev/null +++ b/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() + 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 tag(type: KClass): T? = error("unexpected") + + override fun tag(type: Class): T? = error("unexpected") + + override fun tag( + type: KClass, + computeIfAbsent: () -> T, + ): T = error("unexpected") + + override fun tag( + type: Class, + computeIfAbsent: () -> T, + ): T = error("unexpected") + + override fun clone(): Call = error("unexpected") + + companion object { + val eventListenerUpdater: AtomicReferenceFieldUpdater = + AtomicReferenceFieldUpdater.newUpdater( + RecordingCall::class.java, + EventListener::class.java, + "eventListener", + ) + } + } +} diff --git a/okhttp/src/jvmTest/kotlin/okhttp3/EventListenerTest.kt b/okhttp/src/jvmTest/kotlin/okhttp3/EventListenerTest.kt index d987762ffd46..4c895e7f8237 100644 --- a/okhttp/src/jvmTest/kotlin/okhttp3/EventListenerTest.kt +++ b/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,6 +420,19 @@ 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 @@ -426,12 +440,14 @@ class EventListenerTest( .url(server.url("/")) .build(), ) + val done = CountDownLatch(1) call.enqueue( object : Callback { override fun onFailure( call: Call, e: IOException, ) { + done.countDown() } override fun onResponse( @@ -439,10 +455,16 @@ class EventListenerTest( 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) }