Address connection races (#6867)

* Scope socket state to the connection it belongs to

The socket kept the state of its current transport in its own fields,
so asynchronous events, timers and user callbacks belonging to one
connection could act on the next one, which caused a series of races.

Wrap each transport instance in a Connection, which owns the state of
that connection: its events are only handled while it is the socket's
current connection, and it records whether we closed it, whether the
socket gave up on it, whether its channels were errored, and which
channel joins it carried. Code continuing after user callbacks or
asynchronous waits checks whether the socket still uses its connection,
as callbacks can disconnect or connect at any time.

This fixes that:

- channels stayed joined when the close event arrived late or after
  connect() replaced the connection, resulting in "unmatched topic"
- events of a replaced connection acted on its replacement
- heartbeats kept running after disconnecting, or were restarted by a
  late reply, and then reconnected the socket
- teardowns, heartbeat timeouts and close or error handling reconnected
  after callbacks had disconnected or connected
- disconnecting while connecting fell back to LongPoll, fallback
  callbacks accumulated and acted on later attempts, and a normal close
  by the server did not cancel the fallback health check
- connecting twice while disconnecting leaked a connection
- a stale buffered join was sent in addition to the rejoin
- a heartbeat timeout errored channels while still connected
- resume and visibilitychange tore down the connection they created
- callbacks removed while dispatching a message were skipped
- a normal close by the server was reconnected on visibility change,
  while connect() did nothing
- channels joined by error callbacks were errored again
- asynchronous encoders and decoders acted on replaced or closed
  connections

* Cancel pending LongPoll work when it stops

LongPoll aborted its requests when closing, but aborting can invoke the
request callbacks, synchronously for XHR and asynchronously for fetch,
and those still acted on the closed transport: a spurious error and a
retry, or making it active again after a disconnect. Messages queued
for delivery were still delivered after the close, a retry overrode a
close callback that closed it for good, and an open callback closing it
did not stop the next poll.

A request now belongs to the transport until it completes or the
transport stops, and callbacks of a request that does not belong to it
anymore are ignored. Stopping also cancels the queued messages and the
pending outgoing batch, a retry becomes active again before the close
callbacks run, and polling stops once the transport is closed.

* Pass the channel along with its pushes

Socket.push only received the message, so finding out whether a push
was outdated meant searching all channels for the one with its topic
and join ref, up to three times per push: when it was buffered and
flushed, when it was sent, and when its encoding finished.

Push now passes its channel to socket.push, which turns these checks
into comparing the push's join ref with the channel's current one and
checking whether the channel left.

* fmt

* Make socket tests catch the regressions they are meant for

Breaking the code each test guards showed that some tests no longer
failed for their regression:

- "does not throw when encoding finishes after disconnect completed"
  could not fail anymore, as its join push is dropped as outdated before
  it is sent. It now checks that a push is not sent once its connection
  was replaced while it was encoded.
- "does not connect twice" only checked after the scheduled reconnect,
  when an extra connection was still being torn down. It now also
  checks right after connect().
- "ignores errors and messages of a connection that was replaced"
  closed the old connection, which already detached its error handler.
  The old connection now still drains its buffer.
- "does not error channels when the connection never opened" now
  checks that the channel is not errored at all.

Also cover that replacing the transport while a teardown waits prevents
its reconnect, like disconnecting does.

* Cover dropping the pending LongPoll batch on retry

When LongPoll stops, it now also drops a batch that was still collecting
sends, as its timer is cancelled. Otherwise, sends after a retry were
appended to that batch and never sent.

Also remove the test for a timeout after close, as the tests for abort
callbacks after close and for timeouts after a retry cover it.

* Remove detaching the handlers of a torn down connection

A connection only handles the events of its transport while it is the
socket's current connection, so detaching the handlers of a replaced or
torn down connection made no difference.

* Cover cancelling a scheduled reconnect on a normal close

A normal close by the server cancels a scheduled reconnect, which would
otherwise tear down a connection created by calling connect() again
before the reconnect was due.

* Error channels after socket error callbacks replaced the connection

onConnError skipped erroring the channels when its callbacks replaced
the connection. However, when a callback replaced the transport of a
connection that already failed (and was therefore no longer open),
replaceTransport did not error the channels either, so they stayed
joined and did not rejoin on the next connection.

Always error the channels for the connection afterwards, which only
errors the joins it carried once it was replaced.

* Remove redundant checks of the fallback attempt

fallback() checked whether its attempt was still current, but the
fallback timer is cleared whenever the attempt ends, and the error and
open callbacks only call it while the attempt is current. The ping of
the health check checked it as well, but its reply can only arrive on
the connection that sent it, and it shows that the primary transport
works even after a disconnect.

* more fixes

* longpoll wasClean fix

* don't accumulate ping callbacks when connection drops

Author: SteffenDE

assets/js/phoenix/connection.js

@@ -0,0 +1,73 @@
+import {SOCKET_STATES} from "./constants"
+
+/**
+ * @private
+ *
+ * A connection of the socket over a single transport instance.
+ *
+ * It owns everything that belongs to this connection, so that nothing about it can
+ * affect a later connection: the transport events are only handled while it is the
+ * socket's current connection, and its state is gone together with it.
+ */
+export default class Connection {
+  /**
+   * @param {Socket} socket
+   * @param {Object} transport - The transport instance, for example a WebSocket
+   */
+  constructor(socket, transport){
+    this.socket = socket
+    this.transport = transport
+    // the join ref of each channel whose join was sent over this connection, weakly held
+    // as a connection can outlive many channels, e.g. when navigating between LiveViews
+    this.joins = new WeakMap()
+    // the callbacks of the pings sent over this connection by ref, as their replies can only
+    // arrive over it, so they are gone together with it instead of accumulating on the socket
+    this.pings = new Map()
+    // set when we close the connection, as opposed to the server or the network
+    this.closing = false
+    // set when the socket does not want this connection anymore, e.g. on disconnect,
+    // so that connecting again creates a new one instead of waiting for it
+    this.ended = false
+    // set once the channels were errored after the connection stopped being open
+    this.channelsErrored = false
+    // set once the close event was handled, as LongPoll can emit it more than once
+    this.closeHandled = false
+    transport.onopen = () => { if(this.isCurrent()){ socket.onConnOpen() } }
+    transport.onerror = error => { if(this.isCurrent()){ socket.onConnError(error) } }
+    transport.onmessage = event => { if(this.isCurrent()){ socket.onConnMessage(event) } }
+    transport.onclose = event => {
+      if(this.isCurrent() && !this.closeHandled){
+        this.closeHandled = true
+        socket.onConnClose(event)
+      }
+    }
+  }
+
+  /**
+   * Whether this is the connection the socket currently uses. A replaced connection
+   * can still emit events while it closes, but they no longer concern the socket.
+   */
+  isCurrent(){ return this.socket.connection === this }
+
+  isOpen(){ return this.transport.readyState === SOCKET_STATES.open }
+
+  isConnecting(){ return this.transport.readyState === SOCKET_STATES.connecting }
+
+  /**
+   * Whether the channel's current join was sent over this connection.
+   */
+  carried(channel, joinRef = channel.joinRef()){
+    return this.joins.has(channel) && this.joins.get(channel) === joinRef
+  }
+
+  /**
+   * Closes the transport. Its close event then only confirms our close.
+   */
+  close(code, reason){
+    this.closing = true
+    // closing a transport that is still connecting makes it emit an error, which is
+    // caused by closing it and must not be handled as a transport error
+    this.transport.onerror = function (){ } // noop
+    if(code){ this.transport.close(code, reason || "") } else { this.transport.close() }
+  }
+}

assets/js/phoenix/longpoll.js

@@ -26,7 +26,9 @@ export default class LongPoll {
     this.endPoint = null
     this.token = null
     this.skipHeartbeat = true
+    // the pending requests and queued messages, which are cancelled when we stop
     this.reqs = new Set()
+    this.messageTimers = new Set()
     this.awaitingBatchAck = false
     this.currentBatch = null
     this.currentBatchTimer = null
@@ -37,7 +39,8 @@ export default class LongPoll {
     this.onclose = function (){ } // noop
     this.pollEndpoint = this.normalizeEndpoint(endPoint)
     this.readyState = SOCKET_STATES.connecting
-    // we must wait for the caller to finish setting up our callbacks and timeout properties
+    // we must wait for the caller to finish setting up our callbacks and timeout properties,
+    // and it might close us before that, e.g. when disconnecting right after connecting
     setTimeout(() => this.poll(), 0)
   }
 
@@ -52,19 +55,26 @@ export default class LongPoll {
     return this.pollEndpoint
   }
 
-  closeAndRetry(code, reason, wasClean){
-    this.close(code, reason, wasClean)
-    this.readyState = SOCKET_STATES.connecting
+  closeAndRetry(error, code, reason, wasClean){
+    if(!this.isActive()){ return }
+    // like a failed WebSocket, we are no longer open when we emit the error and then the close.
+    // We are connecting again before their callbacks run, so that they can still close us for
+    // good, e.g. when they disconnect the socket
+    this.stop(SOCKET_STATES.connecting)
+    this.onerror(error)
+    if(this.isActive()){ this.emitClose(code, reason, wasClean) }
   }
 
   ontimeout(){
-    this.onerror("timeout")
-    this.closeAndRetry(1005, "timeout", false)
+    this.closeAndRetry("timeout", 1005, "timeout", false)
   }
 
   isActive(){ return this.readyState === SOCKET_STATES.open || this.readyState === SOCKET_STATES.connecting }
 
   poll(){
+    // onopen can close us before the next poll, just as the caller can close us
+    // before the constructor's deferred first poll
+    if(!this.isActive()){ return }
     const headers = {"Accept": "application/json"}
     if(this.authToken){
       headers["X-Phoenix-AuthToken"] = this.authToken
@@ -75,8 +85,7 @@ export default class LongPoll {
         if(status === 410 && this.token !== null){
           // In case we already have a token, this means that our existing session
           // is gone. We fail so that the client rejoins its channels.
-          this.onerror(410)
-          this.closeAndRetry(3410, "session_gone", false)
+          this.closeAndRetry(410, 3410, "session_gone", false)
           return
         }
         this.token = token
@@ -105,7 +114,14 @@ export default class LongPoll {
             //
             // In order to emulate this behaviour, we need to make sure each
             // onmessage handler is run within its own macrotask.
-            setTimeout(() => this.onmessage({data: msg}), 0)
+            //
+            // We might stop before that, e.g. when a POST times out or a message
+            // handler disconnects, which cancels the remaining messages.
+            const timer = setTimeout(() => {
+              this.messageTimers.delete(timer)
+              this.onmessage({data: msg})
+            }, 0)
+            this.messageTimers.add(timer)
           })
           this.poll()
           break
@@ -123,8 +139,7 @@ export default class LongPoll {
           break
         case 0:
         case 500:
-          this.onerror(500)
-          this.closeAndRetry(1011, "internal server error", 500)
+          this.closeAndRetry(500, 1011, "internal server error", false)
           break
         default: throw new Error(`unhandled poll status ${status}`)
       }
@@ -156,9 +171,8 @@ export default class LongPoll {
     const batch = messages.slice(offset, next)
     this.ajax("POST", {"Content-Type": "application/x-ndjson"}, batch.join("\n"), () => this.ontimeout(), resp => {
       if(!resp || resp.status !== 200){
-        this.awaitingBatchAck = false
-        this.onerror(resp && resp.status)
-        this.closeAndRetry(1011, "internal server error", false)
+        // this calls stop which already clears awaitingBatchAck
+        this.closeAndRetry(resp && resp.status, 1011, "internal server error", false)
       } else if(next < messages.length){
         this.batchSend(messages, next)
       } else if(this.batchBuffer.length > 0){
@@ -171,13 +185,35 @@ export default class LongPoll {
   }
 
   close(code, reason, wasClean){
-    for(let req of this.reqs){ req.abort() }
-    this.readyState = SOCKET_STATES.closed
-    let opts = Object.assign({code: 1000, reason: undefined, wasClean: true}, {code, reason, wasClean})
+    this.stop(SOCKET_STATES.closed)
+    this.emitClose(code, reason, wasClean)
+  }
+
+  /**
+   * Cancels the pending requests, queued messages and outgoing batches.
+   */
+  stop(readyState){
+    this.readyState = readyState
+    // the requests are no longer ours before we abort them, as aborting can invoke their
+    // callbacks, synchronously for XHR and asynchronously for fetch
+    const reqs = this.reqs
+    this.reqs = new Set()
+    for(let {req} of reqs){ req && req.abort() }
+    for(let timer of this.messageTimers){ clearTimeout(timer) }
+    this.messageTimers.clear()
     this.batchBuffer = []
     this.awaitingBatchAck = false
     clearTimeout(this.currentBatchTimer)
     this.currentBatchTimer = null
+    this.currentBatch = null
+  }
+
+  /**
+   * Emits the close event. Like a WebSocket closed without a code, it reports 1005 (no status)
+   * unless one is given, as 1000 would look like the server closing the connection normally.
+   */
+  emitClose(code = 1005, reason, wasClean = true){
+    let opts = {code, reason, wasClean}
     if(typeof(CloseEvent) !== "undefined"){
       this.onclose(new CloseEvent("close", opts))
     } else {
@@ -186,18 +222,19 @@ export default class LongPoll {
   }
 
   ajax(method, headers, body, onCallerTimeout, callback){
-    let req
+    // the request is ours until it completes or we stop, and the callbacks of a request
+    // that is not ours anymore are ignored. It is ours before it starts, as a request can
+    // complete while it starts.
+    const request = {req: null}
+    this.reqs.add(request)
     let ontimeout = () => {
-      this.reqs.delete(req)
-      onCallerTimeout()
+      if(this.reqs.delete(request)){ onCallerTimeout() }
     }
     if(this.token !== null){
       headers = Object.assign({}, headers, {"X-Phoenix-Longpoll-Token": this.token})
     }
-    req = Ajax.request(method, this.endpointURL(), headers, body, this.timeout, ontimeout, resp => {
-      this.reqs.delete(req)
-      if(this.isActive()){ callback(resp) }
+    request.req = Ajax.request(method, this.endpointURL(), headers, body, this.timeout, ontimeout, resp => {
+      if(this.reqs.delete(request)){ callback(resp) }
     })
-    this.reqs.add(req)
   }
 }

assets/js/phoenix/push.js

@@ -40,7 +40,7 @@ export default class Push {
       payload: this.payload(),
       ref: this.ref,
       join_ref: this.channel.joinRef()
-    })
+    }, this.channel)
   }
 
   /**