Fall back to longpoll when the websocket keeps dying after opening

assets/js/phoenix/constants.js

@@ -6,6 +6,9 @@ export const SOCKET_STATES = {connecting: 0, open: 1, closing: 2, closed: 3}
 export const MAX_LONGPOLL_BATCH_SIZE = 100;
 export const DEFAULT_TIMEOUT = 10000
 export const WS_CLOSE_NORMAL = 1000
+// how many primary transport connections may open and then close again without
+// ever delivering a message before we fall back to the fallback transport
+export const MAX_UNUSABLE_PRIMARY_CONNECTIONS = 3
 export const CHANNEL_STATES = {
   closed: "closed",
   errored: "errored",

assets/js/phoenix/socket.js

@@ -4,6 +4,7 @@ import {
   CHANNEL_EVENTS,
   DEFAULT_TIMEOUT,
   DEFAULT_VSN,
+  MAX_UNUSABLE_PRIMARY_CONNECTIONS,
   SOCKET_STATES,
   TRANSPORTS,
   WS_CLOSE_NORMAL,
@@ -120,6 +121,8 @@ export default class Socket {
     this.timeout = opts.timeout || DEFAULT_TIMEOUT
     this.transport = opts.transport || global.WebSocket || LongPoll
     this.primaryPassedHealthCheck = false
+    this.unusablePrimaryConnections = 0
+    this.connReceivedMessage = false
     this.longPollFallbackMs = opts.longPollFallbackMs
     this.fallbackTimer = null
     this.sessionStore = opts.sessionStorage || (global && global.sessionStorage)
@@ -218,6 +221,7 @@ export default class Socket {
   replaceTransport(newTransport){
     this.connectClock++
     this.closeWasClean = true
+    this.unusablePrimaryConnections = 0
     clearTimeout(this.fallbackTimer)
     this.reconnectTimer.reset()
     if(this.conn){
@@ -392,6 +396,7 @@ export default class Socket {
   transportConnect(){
     this.connectClock++
     this.closeWasClean = false
+    this.connReceivedMessage = false
     let protocols = undefined
     // Sec-WebSocket-Protocol based token
     // (longpoll uses Authorization header instead)
@@ -425,6 +430,15 @@ export default class Socket {
       this.transportConnect()
     }
     if(this.getSession(`phx:fallback:${fallbackTransportName}`)){ return fallback("memorized") }
+    // Some networks let the websocket handshake through and then kill the connection
+    // before it ever carries a message. Such a connection counts as established, so
+    // the error handler below never falls back, and as we run again on every
+    // reconnect, the fallback timer is cleared before it can fire. We therefore give
+    // up on the primary transport once enough connections in a row proved unusable.
+    // The fallback is not memorized, because the primary transport did connect.
+    if(this.unusablePrimaryConnections >= MAX_UNUSABLE_PRIMARY_CONNECTIONS){
+      return fallback("unusable primary transport")
+    }
 
     this.fallbackTimer = setTimeout(fallback, fallbackThreshold)
 
@@ -550,6 +564,13 @@ export default class Socket {
     this.triggerChanError("connection_closed")
     this.clearHeartbeats()
     if(!this.closeWasClean && closeCode !== 1000){
+      // a connection that never delivered a message did not prove useful,
+      // which `connectWithFallback` uses to detect a broken primary transport
+      if(this.connReceivedMessage){
+        this.unusablePrimaryConnections = 0
+      } else {
+        this.unusablePrimaryConnections++
+      }
       this.reconnectTimer.scheduleTimeout()
     }
     this.stateChangeCallbacks.close.forEach(([, callback]) => callback(event))
@@ -677,6 +698,7 @@ export default class Socket {
   }
 
   onConnMessage(rawMessage){
+    this.connReceivedMessage = true
     this.decode(rawMessage.data, msg => {
       let {topic, event, payload, ref, join_ref} = msg
       if(ref && ref === this.pendingHeartbeatRef){

assets/test/socket_test.js

@@ -2,7 +2,7 @@ import {jest} from "@jest/globals"
 import {WebSocket, Server as WebSocketServer} from "mock-socket"
 import {encode} from "./serializer"
 import {Socket, LongPoll} from "../js/phoenix"
-import {AUTH_TOKEN_PREFIX, SOCKET_STATES} from "../js/phoenix/constants"
+import {AUTH_TOKEN_PREFIX, MAX_UNUSABLE_PRIMARY_CONNECTIONS, SOCKET_STATES} from "../js/phoenix/constants"
 
 let socket
 
@@ -85,6 +85,24 @@ describe("with transports", function (){
     })
 
     describe("longPollFallbackMs", function (){
+      // a transport that opens and then dies again, optionally delivering a message
+      // before it does, like a middlebox that kills established websockets
+      const dyingTransport = (connections, message) => class DyingTransport {
+        constructor(){
+          this.readyState = SOCKET_STATES.open
+          this.bufferedAmount = 0
+          connections.push(this)
+          setTimeout(() => this.onopen(), 0)
+          if(message){ setTimeout(() => this.onmessage({data: encode(message)}), 5) }
+          setTimeout(() => {
+            this.readyState = SOCKET_STATES.closed
+            this.onclose({code: 1006})
+          }, 10)
+        }
+        close(){ this.readyState = SOCKET_STATES.closed }
+        send(){}
+      }
+
       it("falls back to longpoll when set after primary transport failure", function (done){
         let mockServer
         socket = new Socket("/socket", {longPollFallbackMs: 20})
@@ -101,6 +119,56 @@ describe("with transports", function (){
           socket.connect()
         })
       })
+
+      it("falls back to longpoll when the primary transport keeps dying after opening", function (){
+        jest.useFakeTimers()
+
+        try {
+          const connections = []
+          socket = new Socket("/socket", {
+            transport: dyingTransport(connections),
+            longPollFallbackMs: 2500,
+            reconnectAfterMs: () => 10
+          })
+          const replaceSpy = jest.spyOn(socket, "replaceTransport")
+
+          socket.connect()
+          jest.advanceTimersByTime(200)
+
+          expect(connections.length).toBe(MAX_UNUSABLE_PRIMARY_CONNECTIONS)
+          expect(replaceSpy).toHaveBeenCalledWith(LongPoll)
+          expect(socket.transport).toBe(LongPoll)
+        } finally {
+          jest.useRealTimers()
+        }
+      })
+
+      it("does not fall back when the primary transport delivers messages", function (){
+        jest.useFakeTimers()
+
+        try {
+          const connections = []
+          const transport = dyingTransport(connections, {
+            topic: "phoenix", event: "phx_reply", payload: {status: "ok", response: {}}, ref: "1"
+          })
+          socket = new Socket("/socket", {
+            transport: transport,
+            longPollFallbackMs: 2500,
+            reconnectAfterMs: () => 10
+          })
+          const replaceSpy = jest.spyOn(socket, "replaceTransport")
+
+          socket.connect()
+          jest.advanceTimersByTime(200)
+
+          expect(connections.length).toBeGreaterThan(MAX_UNUSABLE_PRIMARY_CONNECTIONS)
+          expect(replaceSpy).not.toHaveBeenCalled()
+          expect(socket.transport).toBe(transport)
+          expect(socket.unusablePrimaryConnections).toBe(0)
+        } finally {
+          jest.useRealTimers()
+        }
+      })
     })
   })