fix: prevent deadlock on concurrent push and disconnect (#24215) (CP: 25.1) (#24227)

This PR cherry-picks changes from the original PR #24215 to branch 25.1.
---
#### Original PR description
> The previous `AtomicBoolean` guard in `AtmospherePushConnection`
closed only the disconnect-vs-disconnect race. A push thread that reads
`disconnecting` as false before a concurrent `disconnect()` flips it can
still proceed into `synchronized(lock)` behind the disconnect thread,
which is itself blocked inside `resource.close()` waiting for the
servlet container's HTTP session lock held by the push thread — a
two-lock cycle.
> 
> Move `resource.close()` out of the monitor: inside
`synchronized(lock)` capture the resource into a local and call
`connectionLost()` to transition the state, then release the monitor
before invoking `close()` on the stashed reference. Add a matching
re-check of `isConnected()` at the top of the `synchronized` block in
`push()` so a push that waited for the monitor observes the late
disconnect and defers via `PUSH_PENDING`/`RESPONSE_PENDING` instead of
NPEing on the cleared resource. The `disconnecting` flag stays set until
`close()` returns so subsequent pushes take the fast path and no new
`disconnect()` re-enters while `close()` is still in flight.
> 
> Related-to #24192

Co-authored-by: Marco Collovati <marco@vaadin.com>

Author: vaadin-bot

flow-server/src/main/java/com/vaadin/flow/server/communication/AtmospherePushConnection.java

@@ -203,6 +203,20 @@ public void push(boolean async) {
             }
         } else {
             synchronized (lock) {
+                // A concurrent disconnect() may have cleared the
+                // resource and transitioned state to DISCONNECTED while
+                // this thread was waiting to enter the monitor. In that
+                // case treat the push as if disconnecting had been
+                // observed above: defer it and skip sendMessage, which
+                // would otherwise NPE on the null resource.
+                if (!isConnected()) {
+                    if (async && state != State.RESPONSE_PENDING) {
+                        state = State.PUSH_PENDING;
+                    } else {
+                        state = State.RESPONSE_PENDING;
+                    }
+                    return;
+                }
                 try {
                     JsonNode response = new UidlWriter().createUidl(getUI(),
                             async);
@@ -337,15 +351,16 @@ public void disconnect() {
             return;
         }
 
-        synchronized (lock) {
-            if (!isConnected() || resource == null) {
-                // Already disconnected. Should not happen but if it does,
-                // we don't want to cause NPEs
-                getLogger().debug(
-                        "Disconnection already happened, ignoring request");
-                return;
-            }
-            try {
+        AtmosphereResource resourceToClose;
+        try {
+            synchronized (lock) {
+                if (!isConnected() || resource == null) {
+                    // Already disconnected. Should not happen but if it does,
+                    // we don't want to cause NPEs
+                    getLogger().debug(
+                            "Disconnection already happened, ignoring request");
+                    return;
+                }
                 if (resource.isResumed()) {
                     // This can happen for long polling because of
                     // http://dev.vaadin.com/ticket/16919
@@ -374,15 +389,27 @@ public void disconnect() {
                     }
                     outgoingMessage = null;
                 }
+                // Capture the resource and transition this connection to
+                // the disconnected state BEFORE releasing the monitor, so
+                // concurrent push() callers observe an already-closed
+                // connection and skip sendMessage. The actual
+                // resource.close() call is performed outside of
+                // synchronized(lock) below, to avoid a lock-ordering
+                // deadlock between this monitor and the HTTP session
+                // lock that the servlet container may acquire while
+                // closing the AtmosphereResource.
+                resourceToClose = resource;
+                connectionLost();
+            }
+            if (resourceToClose != null) {
                 try {
-                    resource.close();
+                    resourceToClose.close();
                 } catch (IOException e) {
                     getLogger().info("Error when closing push connection", e);
                 }
-                connectionLost();
-            } finally {
-                disconnecting.set(false);
             }
+        } finally {
+            disconnecting.set(false);
         }
     }
 

flow-server/src/test/java/com/vaadin/flow/server/communication/AtmospherePushConnectionTest.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.atmosphere.cpr.AtmosphereResource;
@@ -310,4 +311,131 @@ void pushWhileDisconnect_preventDeadlocks() throws Exception {
         Mockito.verify(resource, Mockito.times(1)).close();
     }
 
+    @Test
+    void pushInterleavedWithDisconnect_preventDeadlocks() throws Exception {
+        // Same motivation as pushWhileDisconnect_preventDeadlocks, but
+        // exercises the race where push() has already read
+        // disconnecting=false BEFORE a concurrent disconnect() flips it
+        // to true. The AtomicBoolean guard does not protect against this
+        // interleaving: disconnect() enters synchronized(lock) and
+        // blocks in resource.close() waiting for the HTTP session lock
+        // held by the push thread, while the push thread then blocks
+        // trying to enter synchronized(lock) held by disconnect(),
+        // producing a deadlock.
+        ReentrantLock httpSessionLock = new ReentrantLock();
+        CountDownLatch disconnectReachedClose = new CountDownLatch(1);
+        Mockito.doAnswer(i -> {
+            // Signal that disconnect() has entered synchronized(lock)
+            // and is about to contend for the HTTP session lock.
+            disconnectReachedClose.countDown();
+            // simulate HTTP session lock attempt because resource.close
+            // accesses session attributes; fail fast if it is still held
+            // by the push thread (indicates a deadlock).
+            if (httpSessionLock.tryLock(2, TimeUnit.SECONDS)) {
+                httpSessionLock.unlock();
+            } else {
+                throw new AssertionError(
+                        "Deadlock on AtmosphereResource.close");
+            }
+            return null;
+        }).when(resource).close();
+
+        CountDownLatch pushReachedBarrier = new CountDownLatch(1);
+        CountDownLatch disconnectProceed = new CountDownLatch(1);
+        AtomicBoolean paused = new AtomicBoolean(false);
+        ThreadLocal<Boolean> pushThreadMarker = ThreadLocal
+                .withInitial(() -> Boolean.FALSE);
+
+        // Pause push() between the disconnecting.get() read and the
+        // synchronized(lock) entry by overriding isConnected(), which is
+        // called in between. Only the push thread's first call pauses,
+        // so a post-fix re-check inside synchronized(lock) does not
+        // re-trigger the hook.
+        UI ui = Mockito.spy(new UI());
+        Mockito.when(ui.getSession()).thenReturn(vaadinSession);
+        AtmospherePushConnection testConnection = new AtmospherePushConnection(
+                ui) {
+            @Override
+            public boolean isConnected() {
+                boolean connected = super.isConnected();
+                if (connected && pushThreadMarker.get()
+                        && paused.compareAndSet(false, true)) {
+                    pushReachedBarrier.countDown();
+                    try {
+                        disconnectProceed.await(2, TimeUnit.SECONDS);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+                return connected;
+            }
+        };
+        testConnection.connect(resource);
+
+        // Dedicated executor to guarantee that push and disconnect can
+        // run concurrently, independent of the common pool parallelism.
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            CompletableFuture<Throwable> pushFuture = CompletableFuture
+                    .supplyAsync(() -> {
+                        pushThreadMarker.set(Boolean.TRUE);
+                        httpSessionLock.lock();
+                        try {
+                            vaadinSession.runWithLock(() -> {
+                                testConnection.push();
+                                return null;
+                            });
+                            return (Throwable) null;
+                        } catch (Throwable t) {
+                            return t;
+                        } finally {
+                            httpSessionLock.unlock();
+                            pushThreadMarker.remove();
+                        }
+                    }, executor);
+
+            // Wait until push() has read disconnecting=false and is
+            // paused just before entering synchronized(lock).
+            assertTrue(pushReachedBarrier.await(2, TimeUnit.SECONDS),
+                    "Push thread did not reach the barrier");
+
+            // Start a concurrent disconnect(). It will CAS disconnecting
+            // from false to true, enter synchronized(lock), and then
+            // attempt resource.close(), which requires the HTTP session
+            // lock held by the push thread.
+            CompletableFuture<Throwable> disconnectFuture = CompletableFuture
+                    .supplyAsync(() -> {
+                        try {
+                            testConnection.disconnect();
+                            return (Throwable) null;
+                        } catch (Throwable t) {
+                            return t;
+                        }
+                    }, executor);
+
+            // Wait deterministically until disconnect() has reached
+            // resource.close() before releasing the push thread.
+            assertTrue(disconnectReachedClose.await(2, TimeUnit.SECONDS),
+                    "Disconnect did not reach resource.close()");
+
+            // Release the push thread so it proceeds toward
+            // synchronized(lock). Without the fix, it blocks here
+            // forever.
+            disconnectProceed.countDown();
+
+            Throwable pushError = pushFuture.get(5, TimeUnit.SECONDS);
+            Throwable disconnectError = disconnectFuture.get(5,
+                    TimeUnit.SECONDS);
+
+            if (disconnectError != null) {
+                fail("Disconnect failed (likely deadlock): " + disconnectError);
+            }
+            if (pushError != null) {
+                fail("Push failed: " + pushError);
+            }
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
 }