fix(ws): serialise session close against in-flight writes (spec-026 US3)

CHANGELOG.md

@@ -6,6 +6,17 @@ The format is inspired by Keep a Changelog, and this project adheres to semantic
 
 ## [Unreleased]
 
+## [2.0.5] - 2026-05-26
+
+### Fixed
+- `NostrRelayClient.send()` no longer leaves the client stuck in the "request in flight" state when the write fails for a reason other than overflow. A plain transport `IOException` from the gated send previously bypassed `pendingRequest` cleanup, so the next `send()` (including an `@NostrRetryable` retry) hit `IllegalStateException: A request is already in flight` instead of retrying. `send()` now clears `pendingRequest` on any send failure, not only `SessionLimitExceededException`. (PR #525 review follow-up.)
+- `closeQuietly()` now logs the swallowed throwable (with stack trace) and the relay URI instead of just the exception message, preserving diagnostics for best-effort closes.
+
+## [2.0.4] - 2026-05-26
+
+### Fixed
+- Close-vs-write race in `NostrRelayClient` (spec-026 US3). The client wrapped its session in Spring's `ConcurrentWebSocketSessionDecorator` but called the **no-arg** `clientSession.close()` (in `close()` and the `send()` timeout path). Spring 6.2.x's decorator overrides only `close(CloseStatus)` (guarded by `closeLock`); the no-arg `close()` falls through to `WebSocketSessionDecorator.close()` → `delegate.close()` with no coordination, sending a CLOSE frame straight to the Tomcat delegate while a `sendMessage` flush was in flight (Tomcat permits only one write in flight → `IllegalStateException: Concurrent write operations are not permitted`). Routing through `close(CloseStatus)` alone is insufficient because its `closeLock` is a separate lock from the `flushLock` guarding `delegate.sendMessage`. Introduced a `ReentrantReadWriteLock` session gate: every write (`send`/`subscribe`) holds the read side (sends stay concurrent — the decorator still serialises the delegate writes among them), every close holds the write side and always uses `close(CloseStatus)`, so a close waits for all in-flight sends to drain before sending the CLOSE frame. No lock nesting → no added deadlock risk.
+
 ### Removed
 - Dead code cleanup — deleted unused classes: `IContent`, `JsonContent`, `Reaction` enum, `Response`, `Nip05Content`, `Nip05ContentDecoder`, `BaseAuthMessage`, `GenericMessage`, `IKey`, `GenericEventConverter`, `GenericEventTypeClassifier`, `GenericEventDecoder`, `FiltersDecoder`, `BaseTagDecoder`, `GenericEventValidator`, `GenericEventSerializer`, `GenericEventUpdater`, `GenericTagQuery`, `HttpClientProvider`, `DefaultHttpClientProvider`.
 - `testAuthMessage` test and `GenericEventSupportTest` removed (tested deleted classes).

nostr-java-client/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>xyz.tcheeric</groupId>
         <artifactId>nostr-java</artifactId>
-        <version>2.0.3</version>
+        <version>2.0.5</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

nostr-java-client/src/main/java/nostr/client/springwebsocket/NostrRelayClient.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Consumer;
 
 /**
@@ -116,6 +117,18 @@ public class NostrRelayClient extends TextWebSocketHandler implements AutoClosea
    */
   private final String relayUri;
   private final ReentrantLock sendLock = new ReentrantLock();
+  /**
+   * Gates session writes against session close. Every {@code clientSession}
+   * write ({@code sendMessage}) holds the READ side; every close holds the
+   * WRITE side. This lets concurrent sends proceed in parallel — the
+   * {@link ConcurrentWebSocketSessionDecorator} still serialises the actual
+   * delegate writes among them — while a close waits for all in-flight sends
+   * to drain before sending a CLOSE frame. Needed because the decorator's
+   * {@code close(CloseStatus)} uses a separate lock from the flush lock, so it
+   * cannot by itself prevent a CLOSE frame from racing an in-flight write on
+   * the Tomcat delegate (which permits only one write in flight).
+   */
+  private final ReentrantReadWriteLock sessionGate = new ReentrantReadWriteLock();
   private PendingRequest pendingRequest;
   private final Map<String, ListenerRegistration> listeners = new ConcurrentHashMap<>();
   private final AtomicReference<ConnectionState> connectionState =
@@ -476,26 +489,29 @@ public List<String> send(String json) throws IOException {
       }
       request = new PendingRequest(maxEventsPerRequest);
       pendingRequest = request;
-      log.info("Sending request to relay {}: {}", relayUri, json);
-      try {
-        clientSession.sendMessage(new TextMessage(json));
-      } catch (SessionLimitExceededException e) {
-        // OverflowStrategy.TERMINATE only sets the limitExceeded flag and throws;
-        // it does NOT close the delegate session. Close it explicitly here so
-        // upstream callers' isOpen()==false reconnect contract holds.
-        pendingRequest = null;
-        try {
-          clientSession.close(CloseStatus.SESSION_NOT_RELIABLE);
-        } catch (IOException closeEx) {
-          // Logged but not propagated — the original cause is the signal.
-          log.warn("Failed to close session after overflow: {}", closeEx.getMessage());
-        }
-        throw new IOException("Failed to send relay payload", e);
-      }
     } finally {
       sendLock.unlock();
     }
 
+    log.info("Sending request to relay {}: {}", relayUri, json);
+    try {
+      sendFrameGated(json);
+    } catch (SessionLimitExceededException e) {
+      // OverflowStrategy.TERMINATE only sets the limitExceeded flag and throws;
+      // it does NOT close the delegate session. Close it explicitly (under the
+      // write gate, outside the read gate just released) so upstream callers'
+      // isOpen()==false reconnect contract holds.
+      clearPendingRequest(request);
+      closeQuietly(CloseStatus.SESSION_NOT_RELIABLE, "after overflow");
+      throw new IOException("Failed to send relay payload", e);
+    } catch (IOException | RuntimeException e) {
+      // Any other send failure must also release the in-flight slot, or the next
+      // send() — including @NostrRetryable's own retry — hits the "request already
+      // in flight" guard instead of retrying.
+      clearPendingRequest(request);
+      throw e;
+    }
+
     long timeout = awaitTimeoutMs > 0 ? awaitTimeoutMs : DEFAULT_AWAIT_TIMEOUT_MS;
     log.debug("Waiting for relay response with timeout={}ms", timeout);
 
@@ -513,11 +529,7 @@ public List<String> send(String json) throws IOException {
       } finally {
         sendLock.unlock();
       }
-      try {
-        clientSession.close();
-      } catch (IOException closeEx) {
-        log.warn("Error closing session after timeout", closeEx);
-      }
+      closeQuietly(CloseStatus.SESSION_NOT_RELIABLE, "after timeout");
       throw new RelayTimeoutException(timeout);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
@@ -595,7 +607,7 @@ public AutoCloseable subscribe(
         new ListenerRegistration(messageListener, errorListener, closeListener));
 
     try {
-      clientSession.sendMessage(new TextMessage(requestJson));
+      sendFrameGated(requestJson);
     } catch (IOException e) {
       listeners.remove(listenerId);
       throw e;
@@ -605,12 +617,7 @@ public AutoCloseable subscribe(
       // it does NOT close the delegate session. Close it explicitly here so
       // upstream callers' isOpen()==false reconnect contract holds.
       if (e instanceof SessionLimitExceededException) {
-        try {
-          clientSession.close(CloseStatus.SESSION_NOT_RELIABLE);
-        } catch (IOException closeEx) {
-          // Logged but not propagated — the original cause is the signal.
-          log.warn("Failed to close session after overflow: {}", closeEx.getMessage());
-        }
+        closeQuietly(CloseStatus.SESSION_NOT_RELIABLE, "after overflow");
       }
       throw new IOException("Failed to send subscription payload", e);
     }
@@ -698,16 +705,78 @@ public AutoCloseable recoverSubscription(
 
   @Override
   public void close() throws IOException {
-    if (clientSession != null) {
+    closeGated(CloseStatus.NORMAL);
+  }
+
+  /**
+   * Write one frame on the delegate while holding the READ side of the session
+   * gate, so it can run concurrently with other sends (the
+   * {@link ConcurrentWebSocketSessionDecorator} serialises the delegate writes
+   * among them) but never overlaps a close (which takes the WRITE side).
+   */
+  private void sendFrameGated(String json) throws IOException {
+    sessionGate.readLock().lock();
+    try {
+      clientSession.sendMessage(new TextMessage(json));
+    } finally {
+      sessionGate.readLock().unlock();
+    }
+  }
+
+  /**
+   * Close the underlying session through {@code close(CloseStatus)} while
+   * holding the WRITE side of the session gate. This excludes every in-flight
+   * {@link #sendFrameGated} (which holds the read side), so the delegate never
+   * sees a CLOSE frame and a data frame at the same time. The no-arg
+   * {@code close()} is deliberately never used: it is not overridden by
+   * {@link ConcurrentWebSocketSessionDecorator} and would bypass its
+   * close/flush coordination entirely.
+   */
+  private void closeGated(CloseStatus status) throws IOException {
+    if (clientSession == null) {
+      return;
+    }
+    sessionGate.writeLock().lock();
+    try {
       boolean open = false;
       try {
         open = clientSession.isOpen();
       } catch (Exception e) {
         log.warn("Exception while checking if clientSession is open during close()", e);
       }
       if (open) {
-        clientSession.close();
+        clientSession.close(status);
+      }
+    } finally {
+      sessionGate.writeLock().unlock();
+    }
+  }
+
+  /**
+   * Best-effort {@link #closeGated} that logs rather than propagates failures —
+   * including the {@code SessionLimitExceededException} the decorator may raise
+   * while closing, so a best-effort close never masks the original timeout or
+   * overflow signal the caller is about to throw.
+   */
+  private void closeQuietly(CloseStatus status, String reason) {
+    try {
+      closeGated(status);
+    } catch (Exception e) {
+      // Swallow but preserve diagnostics — pass the throwable so the stack trace
+      // and root cause are logged, not just the message.
+      log.warn("Error closing session {} on relay {}", reason, relayUri, e);
+    }
+  }
+
+  /** Clear {@link #pendingRequest} if it still points at {@code request}. */
+  private void clearPendingRequest(PendingRequest request) {
+    sendLock.lock();
+    try {
+      if (pendingRequest == request) {
+        pendingRequest = null;
       }
+    } finally {
+      sendLock.unlock();
     }
   }