hugo-vrijswijk · Sep 4, 2026 · Sep 4, 2026
diff --git a/core/src/main/scala/clue/websocket/ApolloClient.scala b/core/src/main/scala/clue/websocket/ApolloClient.scala
@@ -264,6 +264,10 @@ class ApolloClient[F[_], P, S](
   // </WebSocketHandler>
 
   // <ApolloClient Helpers>
+  /** Closes the socket without blocking the caller, which owns another transition. */
+  private def closeInBackground(connection: WebSocketConnection[F]): F[Unit] =
+    connection.closeInternal(none).start.void
+
   private def handleRetry(
     t:                Throwable,
     oldConnection:    Option[WebSocketConnection[F]],
@@ -273,15 +277,15 @@ class ApolloClient[F[_], P, S](
     newLatch:         Latch[F, JsonObject],
     attempt:          Int
   ): (State[F], F[Unit]) = {
-    val disconnectBackend: F[Unit] =
-      oldConnection.map(_.closeInternal(none).start.void).getOrElse(F.unit)
+    val disconnectBackend: F[Unit] = oldConnection.traverse_(closeInBackground)
 
     val errorSubscriptions: F[Unit] = crashSubscriptions(subscriptions, t)
 
     reconnectionStrategy(attempt, t.asLeft) match
       case None       =>
+        // The latch is shared with connect() callers. The error must reach it or a caller waits forever.
         Disconnected(nextConnectionId, t.some) ->
-          (errorSubscriptions >> disconnectBackend >> t.logAndRaiseF)
+          (newLatch.error(t) >> errorSubscriptions >> disconnectBackend >> t.logAndRaiseF)
       case Some(wait) =>
         Connecting(nextConnectionId, none, payload, subscriptions, newLatch) ->
           (t.warnF(s"Error in connect() after attempt #[$attempt]. Retrying.") >>
@@ -298,14 +302,19 @@ class ApolloClient[F[_], P, S](
         .attempt
         .flatMap: connectionAttempt =>
           stateModify:
-            case Connecting(connectionId, None, payload, subscriptions, latch) =>
+            case Connecting(stateConnectionId, None, payload, subscriptions, latch)
+                if connectionId === stateConnectionId =>
               connectionAttempt match
                 case Left(t)           =>
                   handleRetry(t, none, connectionId.next, payload, subscriptions, latch, attempt)
                 case Right(connection) =>
                   Connecting(connectionId, connection.some, payload, subscriptions, latch) ->
-                    doInitialize(connection, payload, latch, attempt)
-            case s                                                             =>
+                    doInitialize(connectionId, connection, payload, latch, attempt)
+            // The state moved to another connection, so this result is stale. Close the socket.
+            case s if s.connectionId =!= connectionId =>
+              s -> (s"Discarding connect() result of a replaced connection.".debugF >>
+                connectionAttempt.toOption.traverse_(closeInBackground))
+            case s                                    =>
               s -> (s"Unexpected state in connect().".errorF >> s"State Is: [$s]".traceF >>
                 InvalidInvocationException(
                   s"Unexpected state in connect(). Unblocking clients, but state may be inconsistent."
@@ -315,10 +324,11 @@ class ApolloClient[F[_], P, S](
           case Outcome.Canceled()                        => disconnect().start.void // Cleanup
 
   private def doInitialize(
-    connection: WebSocketConnection[F],
-    payload:    F[JsonObject],
-    latch:      Latch[F, JsonObject],
-    attempt:    Int
+    connectionId: ConnectionId,
+    connection:   WebSocketConnection[F],
+    payload:      F[JsonObject],
+    latch:        Latch[F, JsonObject],
+    attempt:      Int
   ): F[Unit] =
     (for
       p        <- payload
@@ -327,25 +337,35 @@ class ApolloClient[F[_], P, S](
       result   <- latch.resolve.attempt // Sync up with server response.
       newLatch <- Latch[F, JsonObject]
       _        <- stateModify {
-                    case Connecting(connectionId, Some(connection), payload, subscriptions, _) =>
+                    case Connecting(
+                          stateConnectionId,
+                          Some(stateConnection),
+                          payload,
+                          subscriptions,
+                          _
+                        ) if connectionId === stateConnectionId =>
                       result match
                         case Left(t)  =>
                           handleRetry(
                             t,
-                            connection.some,
+                            stateConnection.some,
                             connectionId.next,
                             payload,
                             subscriptions,
                             newLatch,
                             attempt
                           )
                         case Right(_) =>
-                          Connected(connectionId, connection, payload, subscriptions) ->
-                            startSubscriptions(connection, subscriptions)
-                    case s @ Disconnected(_, _) if result.isLeft                               =>
+                          Connected(connectionId, stateConnection, payload, subscriptions) ->
+                            startSubscriptions(stateConnection, subscriptions)
+                    // The state moved to another connection, so this result is stale.
+                    // Replaced connections usually end in Disconnected state.
+                    case s if s.connectionId =!= connectionId    =>
+                      s -> s"Discarding initialization result of a replaced connection.".debugF
+                    case s @ Disconnected(_, _) if result.isLeft =>
                       s -> (s"Disconnected while initializing.".debugF >>
                         RemoteInitializationException(result.swap.toOption.get).raiseF)
-                    case s                                                                     =>
+                    case s                                       =>
                       s -> (s"Unexpected state when initializing.".errorF >> s"State Is: [$s]".traceF >>
                         InvalidInvocationException(
                           s"Unexpected state when initializing. State may be inconsistent."

diff --git a/core/src/test/scala/clue/websocket/ApolloClientCloseSuite.scala b/core/src/test/scala/clue/websocket/ApolloClientCloseSuite.scala
@@ -117,6 +117,19 @@ final class ApolloClientCloseSuite extends ApolloClientSuite:
       done    <- reader.join.as(true).timeoutTo(Timeout, IO.pure(false))
     yield assert(done, "The subscription stream did not end.")
 
+  test("A connect() call fails when the reconnection gives up"):
+    for
+      backend <- TestWebSocketBackend[IO](autoAck = false)
+      client  <- clientOn(backend, retryOnce, connect = false)
+      // The caller waits for the acknowledgement of the first connection.
+      fiber   <- client.connect().attempt.start
+      _       <- backend.awaitConnectionInits(1).timeout(Timeout)
+      // The reconnection cannot open a socket. The strategy gives up on a connect error.
+      _       <- backend.failConnects(true)
+      _       <- backend.closeFromServer(CloseParams(1000, "retry").asRight).start
+      outcome <- fiber.joinWithNever.timeout(Timeout)
+    yield assertEquals(outcome.leftMap(_.getMessage), "connect failed".asLeft)
+
   test("A subscription that waits for a connection fails with the cause of the close"):
     for
       backend <- TestWebSocketBackend[IO](autoAck = false)

diff --git a/core/src/test/scala/clue/websocket/ApolloClientConnectionIdSuite.scala b/core/src/test/scala/clue/websocket/ApolloClientConnectionIdSuite.scala
@@ -0,0 +1,69 @@
+// Copyright (c) 2016-2025 Association of Universities for Research in Astronomy, Inc. (AURA)
+// For license information see LICENSE or https://opensource.org/licenses/BSD-3-Clause
+
+package clue.websocket
+
+import cats.effect.*
+import cats.syntax.all.*
+    import clue.PersistentClientStatus
+import clue.model.StreamingMessage
+
+/**
+ * Tests for the connection identity of the client state machine. A result that belongs to a
+ * connection that the state replaced must not change the state.
+ */
+final class ApolloClientConnectionIdSuite extends ApolloClientSuite:
+
+  test("A connect result for a replaced connection is dropped"):
+    for
+      backend          <- TestWebSocketBackend[IO](autoAck = false)
+      gates            <- backend.gateConnects(2)
+      (client, logger) <- clientWithLogger(backend, connect = false)
+      // The first connect waits in the backend. The disconnect replaces the connection id.
+      first            <- client.connect().attempt.start
+      _                <- awaitStatus(client, PersistentClientStatus.Connecting).timeout(Timeout)
+      _                <- client.disconnect()
+      second           <- client.connect().start
+      _                <- awaitStatus(client, PersistentClientStatus.Connecting).timeout(Timeout)
+      // The first connect now returns a connection that the client must not use.
+      _                <- gates.head.complete(())
+      _                <- first.joinWithNever.timeout(Timeout)
+      stale            <- backend.sent
+      _                <- IO(assertEquals(stale, Nil, "The client used a replaced connection."))
+      // The socket of the replaced connection must not leak. The client closes it in background.
+      _                <- backend.awaitCloses(1).timeout(Timeout)
+      closed           <- backend.closes
+      _                <- IO(
+                            assertEquals(closed,
+                                         List(Option.empty[CloseParams]),
+                                         "The client did not close the replaced socket."
+                            )
+                          )
+      // The current connect still completes.
+      _                <- gates(1).complete(())
+      _                <- backend.awaitConnectionInits(1).timeout(Timeout)
+      _                <- backend.emit(StreamingMessage.FromServer.ConnectionAck())
+      _                <- second.joinWithNever.timeout(Timeout)
+      status           <- client.status
+      errs             <- errors(logger)
+    yield
+      assertEquals(status, PersistentClientStatus.Connected)
+      assertEquals(errs, Vector.empty)
+
+  test("An initialization result for a replaced connection is dropped"):
+    for
+      backend          <- TestWebSocketBackend[IO](autoAck = false)
+      (client, logger) <- clientWithLogger(backend, retryOnce, connect = false)
+      first            <- client.connect().start
+      _                <- backend.awaitConnectionInits(1).timeout(Timeout)
+      // The reconnection keeps the latch, so two initializations wait for one acknowledgement.
+      _                <- backend.closeFromServer(CloseParams(1000, "retry").asRight).start
+      _                <- backend.awaitConnectionInits(2).timeout(Timeout)
+      _                <- backend.emit(StreamingMessage.FromServer.ConnectionAck())
+      _                <- awaitStatus(client, PersistentClientStatus.Connected).timeout(Timeout)
+      _                <- first.joinWithNever.timeout(Timeout)
+      errs             <- errors(logger)
+    yield assertEquals(errs,
+                       Vector.empty,
+                       "The client reported an error for a replaced connection."
+    )
diff --git a/core/src/test/scala/clue/websocket/ApolloClientSuite.scala b/core/src/test/scala/clue/websocket/ApolloClientSuite.scala
@@ -40,10 +40,11 @@ trait ApolloClientSuite extends CatsEffectSuite:
   /** Builds a client with a logger that the test can inspect. */
   protected def clientWithLogger(
     backend:  TestWebSocketBackend[IO],
-    strategy: ReconnectionStrategy = ReconnectionStrategy.never
+    strategy: ReconnectionStrategy = ReconnectionStrategy.never,
+    connect:  Boolean = true
   ): IO[(ApolloClient[IO, String, Unit], TestingLogger[IO])] =
     val logger = TestingLogger.impl[IO]()
-    clientOn(backend, strategy)(using logger).tupleRight(logger)
+    clientOn(backend, strategy, connect)(using logger).tupleRight(logger)
 
   /** Reconnects on a close with the reason "retry", and gives up on any other. */
   protected val retryOnce: ReconnectionStrategy = (_, reason) =>

diff --git a/core/src/test/scala/clue/websocket/TestWebSocketBackend.scala b/core/src/test/scala/clue/websocket/TestWebSocketBackend.scala
@@ -21,25 +21,45 @@ import io.circe.syntax.*
  * messages that the client sends. It feeds server messages and close events to the client.
  */
 final class TestWebSocketBackend[F[_]: Async] private (
-  sentRef:      SignallingRef[F, Vector[StreamingMessage.FromClient]],
-  currentRef:   Ref[F, Option[(PersistentBackendHandler[F, CloseEvent], ConnectionId)]],
-  closesRef:    Ref[F, Vector[Option[CloseParams]]],
-  autoAckRef:   Ref[F, Boolean],
-  failSendsRef: Ref[F, Boolean]
+  sentRef:         SignallingRef[F, Vector[StreamingMessage.FromClient]],
+  currentRef:      Ref[F, Option[(PersistentBackendHandler[F, CloseEvent], ConnectionId)]],
+  closesRef:       SignallingRef[F, Vector[Option[CloseParams]]],
+  autoAckRef:      Ref[F, Boolean],
+  failSendsRef:    Ref[F, Boolean],
+  failConnectsRef: Ref[F, Boolean],
+  gatesRef:        Ref[F, List[Deferred[F, Unit]]]
 ) extends WebSocketBackend[F, String]:
 
   /** Turns the automatic `connection_ack` answer on or off. */
   def autoAck(enabled: Boolean): F[Unit] = autoAckRef.set(enabled)
 
+  /**
+   * Makes the first `count` connections wait for returned gates. Gate at index `i` belongs to
+   * connection id `i`.
+   */
+  def gateConnects(count: Int): F[List[Deferred[F, Unit]]] =
+    List.fill(count)(Deferred[F, Unit]).sequence.flatTap(gatesRef.set)
+
+  /** Waits for the gate of this connection, if the test set one. */
+  private def awaitGate(connectionId: ConnectionId): F[Unit] =
+    gatesRef.get.flatMap(_.lift(connectionId.value).traverse_(_.get))
+
   /** Makes every further `send` raise, like a socket that died without a close event. */
   def failSends(enabled: Boolean): F[Unit] = failSendsRef.set(enabled)
 
+  /** Makes every further `connect` raise, like a socket that cannot open. */
+  def failConnects(enabled: Boolean): F[Unit] = failConnectsRef.set(enabled)
+
   /** The messages that the client sent, in order. */
   val sent: F[List[StreamingMessage.FromClient]] = sentRef.get.map(_.toList)
 
   /** The close parameters of every `closeInternal` call, in order. */
   val closes: F[List[Option[CloseParams]]] = closesRef.get.map(_.toList)
 
+  /** Waits until the client closed at least `count` connections. */
+  def awaitCloses(count: Int): F[Unit] =
+    closesRef.discrete.find(_.sizeIs >= count).compile.drain
+
   /** The subscribe messages that the client sent, in order. */
   val subscribes: F[List[StreamingMessage.FromClient.Subscribe]] =
     sent.map(_.collect { case s: StreamingMessage.FromClient.Subscribe => s })
@@ -87,8 +107,11 @@ final class TestWebSocketBackend[F[_]: Async] private (
     handler:          PersistentBackendHandler[F, CloseEvent],
     connectionId:     ConnectionId
   ): F[WebSocketConnection[F]] =
-    currentRef
-      .set((handler, connectionId).some)
+    (awaitGate(connectionId) >>
+      failConnectsRef.get.ifM(
+        Async[F].raiseError(new RuntimeException("connect failed")),
+        currentRef.set((handler, connectionId).some)
+      ))
       .as(
         new PersistentConnection[F, CloseParams]:
           override def send(msg: StreamingMessage.FromClient): F[Unit] =
@@ -116,9 +139,11 @@ final class TestWebSocketBackend[F[_]: Async] private (
 object TestWebSocketBackend:
   def apply[F[_]: Async](autoAck: Boolean = true): F[TestWebSocketBackend[F]] =
     for
-      sent   <- SignallingRef.of[F, Vector[StreamingMessage.FromClient]](Vector.empty)
-      cur    <- Ref.of[F, Option[(PersistentBackendHandler[F, CloseEvent], ConnectionId)]](none)
-      closes <- Ref.of[F, Vector[Option[CloseParams]]](Vector.empty)
-      ack    <- Ref.of[F, Boolean](autoAck)
-      fail   <- Ref.of[F, Boolean](false)
-    yield new TestWebSocketBackend(sent, cur, closes, ack, fail)
+      sent        <- SignallingRef.of[F, Vector[StreamingMessage.FromClient]](Vector.empty)
+      cur         <- Ref.of[F, Option[(PersistentBackendHandler[F, CloseEvent], ConnectionId)]](none)
+      closes      <- SignallingRef.of[F, Vector[Option[CloseParams]]](Vector.empty)
+      ack         <- Ref.of[F, Boolean](autoAck)
+      failSend    <- Ref.of[F, Boolean](false)
+      failConnect <- Ref.of[F, Boolean](false)
+      gates       <- Ref.of[F, List[Deferred[F, Unit]]](Nil)
+    yield new TestWebSocketBackend(sent, cur, closes, ack, failSend, failConnect, gates)
diff --git a/sbt-plugin/src/main/scala-2.12/clue/sbt/PluginCompat.scala b/sbt-plugin/src/main/scala-2.12/clue/sbt/PluginCompat.scala
@@ -3,8 +3,8 @@
 
 package clue.sbt
 
-import org.portablescala.sbtplatformdeps.PlatformDepsPlugin.autoImport._
-import sbt._
+import org.portablescala.sbtplatformdeps.PlatformDepsPlugin.autoImport.*
+import sbt.*
 
 /**
  * sbt 1.x half of the compatibility shim. See `src/main/scala-3` for the sbt 2.x half.