Send the fallback health check before the buffered join

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Bug fixes
 
+  * [phoenix.js] Send the fallback health check before the buffered join, and
+    treat any frame on that websocket as a pass
+    ([#6770](https://github.com/phoenixframework/phoenix/issues/6770))
   * [Phoenix.ChannelTest] Fix test sockets violating the Phoenix.Socket.t
     typespec, which made Dialyzer treat every test join as never returning
     ([#5509](https://github.com/phoenixframework/phoenix/issues/5509))

assets/js/phoenix/socket.js

@@ -511,14 +511,9 @@ export default class Socket {
       }
       if(!isCurrent()){ return }
       established = true
-      // if we've established primary, give the fallback a new period to attempt ping
+      // the health check already went out. This period is how long a frame has to arrive.
       clearTimeout(this.fallbackTimer)
       this.fallbackTimer = setTimeout(fallback, fallbackThreshold)
-      this.ping(rtt => {
-        this.log("transport", "connected to primary after", rtt)
-        this.primaryPassedHealthCheck = true
-        clearTimeout(this.fallbackTimer)
-      })
     })
     this.transportConnect()
     connectClock = this.connectClock
@@ -545,12 +540,44 @@ export default class Socket {
     if(this.hasLogger()) this.log("transport", `${this.transportName(this.transport)} connected to ${this.endPointURL()}`)
     this.closeWasClean = false
     this.establishedConnections++
+    // the health-check ping has to leave before the buffered join. The transport
+    // process blocks in Channel.Server.join until the join returns, so a ping
+    // sent after the join is not answered while the mount runs, and the fallback
+    // timer then drops a socket that is fine.
+    this.pingBeforeBufferedJoin()
     this.flushSendBuffer()
     this.reconnectTimer.reset()
     this.resetHeartbeat()
     this.stateChangeCallbacks.open.forEach(([, callback]) => callback())
   }
 
+  /**
+   * @private
+   *
+   * Sends the fallback health check before flushSendBuffer. No-op unless a
+   * fallback attempt is waiting on the primary transport.
+   */
+  pingBeforeBufferedJoin(){
+    if(!this.fallbackTimer || this.transport === LongPoll){ return }
+    this.ping(rtt => {
+      this.log("transport", "connected to primary after", rtt)
+      this.passPrimaryHealthCheck()
+    })
+  }
+
+  /**
+   * @private
+   *
+   * The primary transport answered. Later frames on this connection also call
+   * this, because a frame is the proof the ping was waiting for.
+   */
+  passPrimaryHealthCheck(){
+    if(this.transport === LongPoll){ return }
+    this.primaryPassedHealthCheck = true
+    clearTimeout(this.fallbackTimer)
+    this.fallbackTimer = null
+  }
+
   /**
    * @private
    */
@@ -854,6 +881,9 @@ export default class Socket {
       // connection may have been replaced or closed in the meantime (browsers also drop messages
       // that arrive once a WebSocket is closing)
       if(connection !== this.connection || (wasOpen && !connection.isOpen())){ return }
+      // a frame on this connection is the health check. The ping reply is one
+      // such frame, and so is anything the server already sent.
+      if(this.fallbackTimer){ this.passPrimaryHealthCheck() }
       let {topic, event, payload, ref, join_ref} = msg
       if(ref && ref === this.pendingHeartbeatRef){
         this.clearHeartbeats()

assets/test/socket_test.js

@@ -197,6 +197,51 @@ describe("with transports", function (){
           expect(socket.stateChangeCallbacks.error.length).toBe(1)
         })
 
+        it("sends the health check before the buffered join", function (){
+          const channel = socket.channel("topic")
+          channel.join()
+          socket.connect()
+          connections[0].open()
+
+          expect(connections[0].sent.map(([, , , event]) => event)).toEqual(["heartbeat", "phx_join"])
+        })
+
+        it("does not fall back when the health check is answered before the join reply", function (){
+          const channel = socket.channel("topic")
+          channel.join()
+          socket.connect()
+          connections[0].open()
+          const [[, heartbeatRef]] = connections[0].sent.filter(([, , topic, event]) => topic === "phoenix" && event === "heartbeat")
+          connections[0].onmessage({data: encode({
+            ref: heartbeatRef,
+            topic: "phoenix",
+            event: "phx_reply",
+            payload: {status: "ok", response: {}}
+          })})
+
+          jest.advanceTimersByTime(2500)
+
+          expect(socket.transport).toBe(StubWebSocket)
+          expect(socket.primaryPassedHealthCheck).toBe(true)
+          expect(socket.getSession("phx:fallback:LongPoll")).toBeNull()
+        })
+
+        it("does not fall back when any frame arrives on the open websocket", function (){
+          socket.connect()
+          connections[0].open()
+          connections[0].onmessage({data: encode({
+            ref: null,
+            topic: "phoenix",
+            event: "phx_reply",
+            payload: {status: "ok", response: {}}
+          })})
+
+          jest.advanceTimersByTime(2500)
+
+          expect(socket.transport).toBe(StubWebSocket)
+          expect(socket.primaryPassedHealthCheck).toBe(true)
+        })
+
         it("does not accumulate the health check pings of connections that dropped", function (){
           socket.connect()
           for(let i = 0; i < 3; i++){