Avoid leaked connections being re-allocated

okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/connection/RealConnectionPool.kt

@@ -307,6 +307,16 @@ class RealConnectionPool internal constructor(
 
       // If this was the last allocation, the connection is eligible for immediate eviction.
       if (references.isEmpty()) {
+        if (!connection.isMultiplexed && !connection.noNewExchanges) {
+          // If we have cleared a leaked call reference for a HTTP/1.1 connection,
+          // we must ensure this can't be reallocated even if multiple connections were leaked
+          // and only one is about to be closed
+          connection.noNewExchanges = true
+
+          // Can't notify while holding lock
+          // connection.connectionListener.noNewExchanges(connection)
+        }
+
         connection.idleAtNs = now - keepAliveDurationNs
         return 0
       }

okhttp/src/jvmTest/kotlin/okhttp3/internal/connection/ConnectionPoolTest.kt

@@ -20,6 +20,7 @@ import assertk.assertions.isEmpty
 import assertk.assertions.isEqualTo
 import assertk.assertions.isFalse
 import assertk.assertions.isNotEmpty
+import assertk.assertions.isNotSameInstanceAs
 import assertk.assertions.isTrue
 import okhttp3.ConnectionPool
 import okhttp3.FakeRoutePlanner
@@ -178,6 +179,34 @@ class ConnectionPoolTest {
     assertThat(c1.noNewExchanges).isTrue()
   }
 
+  @Test fun multipleLeakedAllocations() {
+    val pool = factory.newConnectionPool()
+    val poolApi = ConnectionPool(pool)
+    val c1 = factory.newConnection(pool, routeA1, 0L)
+    allocateAndLeakAllocation(poolApi, c1)
+    val c2 = factory.newConnection(pool, routeA1, 0L)
+    allocateAndLeakAllocation(poolApi, c2)
+    awaitGarbageCollection()
+    assertThat(pool.closeConnections(100L)).isEqualTo(0L)
+
+    assertThat(c1.calls).isEmpty()
+    assertThat(c1.noNewExchanges).isTrue()
+
+    assertThat(c2.calls).isEmpty()
+    assertThat(c2.noNewExchanges).isTrue()
+
+    val client =
+      OkHttpClient
+        .Builder()
+        .connectionPool(poolApi)
+        .build()
+    val call = client.newCall(Request(c1.route().address.url)) as RealCall
+    val c3 = pool.callAcquirePooledConnection(false, c1.route().address, call, null, false)
+
+    assertThat(c3).isNotSameInstanceAs(c1)
+    assertThat(c3).isNotSameInstanceAs(c2)
+  }
+
   @Test fun interruptStopsThread() {
     val taskRunnerThreads = mutableListOf<Thread>()
     val taskRunner =