[fix] [#417] Fix broken server->client broadcast on client IP change (@…

…Naomarik)

Huge thanks to @Naomarik for reporting and diagnosing this issue!

BEFORE THIS COMMIT
  The following scenario was possible:

  t0. Client WebSocket connects to server with unique client ID.
      State in `conns_` atom: `{<client-id1> [<ws-sch:port-for-ip1> <udt>]}`.

  t1. Client's IP changes (for example, by switching from wifi to cellular network).

  t2. No `:on-close` event is triggered on server.

  t3. New client WebSocket connects to server with the SAME client ID but new IP.
      State in `conns_` atom: `{<client-id1> [<ws-sch:port-for-ip2> <udt>]}`.

  t4. Server-side connection timeout triggers for ORIGINAL connection (IP1).
      This unintentionally removes sch for the NEW connection (IP2).
      State in `conns_` atom: `{<client-id1> [nil <udt>]}`.

  t5. At this point:
    - The client has a working WebSocket connection to server.
    - But the server `conns_` state is bad (has a nil sch for client).
    - Which means that server->client broadcasts all fail.
    - This broken state persists until if/when something causes the client
      to reconnect. But note that this may not happen anytime soon since
      the client believes (accurately) that it IS successfully connected.

IMPLEMENTATION DETAILS

   Each connection to server establishes the following:
     - An `:on-open`  handler that modifies state in `conns_` for <client-id>
     - An `:on-close` handler that modifies state in `conns_` for <client-id>

   The bad behaviour occurs when:

     - Conn1's delayed `:on-close` triggers after Conn2's `:on-open`.
       Because Conn1 and Conn2 share the same client-id, they're mutating
       the same state.

AFTER THIS COMMIT

  We introduce a simple compare-and-swap (CAS) mechanism in the `:on-close`
  handler so that its state mutations will noop if the current server-ch does
  not = the server-ch added by the corresponding `:on-open` handler.

  I.e. a given `:on-close` will now only remove the SAME server-ch added by
  its corresponding `:on-open`.

  In the example above, this means that the timeout at t4 will noop.

src/taoensso/sente.cljc

@@ -457,21 +457,27 @@
         send-buffers_   (atom {:ws  {} :ajax  {}}) ; {<uid> [<buffered-evs> <#{ev-uuids}>]}
         connected-uids_ (atom {:ws #{} :ajax #{} :any #{}}) ; Public
 
-        upd-conn!
+        upd-conn! ; Update client entry in conns_
         (fn
-          ([conn-type uid client-id] ; Update udt
+          ([conn-type uid client-id] ; Update only udt
            (swap-in! conns_ [conn-type uid client-id]
              (fn [?v]
-               (let [[?sch _udt] ?v
+               (let [[?sch _?udt] ?v
                      new-udt (enc/now-udt)]
                  (enc/swapped
                    [?sch new-udt]
                    {:init? (nil? ?v) :udt new-udt :?sch ?sch})))))
 
-          ([conn-type uid client-id new-?sch] ; Update sch + udt
+          ([conn-type uid client-id old-?sch new-?sch] ; Update sch & udt
            (swap-in! conns_ [conn-type uid client-id]
              (fn [?v]
-               (let [new-udt (enc/now-udt)]
+               (let [[?sch _?udt] ?v
+                     new-udt (enc/now-udt)
+                     new-?sch
+                     (if (or (= old-?sch :any) (identical? old-?sch ?sch))
+                       new-?sch ; CAS successful, Ref. #417
+                       ?sch)]
+
                  (enc/swapped
                    [new-?sch new-udt]
                    {:init? (nil? ?v) :udt new-udt :?sch new-?sch}))))))
@@ -708,6 +714,7 @@
              params     (get ring-req :params)
              client-id  (get params   :client-id)
              uid        (user-id-fn    ring-req client-id)
+             ;; ?ws-key (get-in ring-req [:headers "sec-websocket-key"])
 
              receive-event-msg! ; Partial
              (fn self
@@ -748,7 +755,7 @@
 
                   ;; WebSocket handshake
                   (let [_ (tracef "New WebSocket channel: %s (%s)" uid sch-uuid)
-                        updated-conn (upd-conn! :ws uid client-id server-ch)
+                        updated-conn (upd-conn! :ws uid client-id :any server-ch)
                         udt-open     (:udt updated-conn)]
 
                     (when (connect-uid! :ws uid)
@@ -774,7 +781,7 @@
 
                   ;; Ajax handshake/poll
                   (let [_ (tracef "New Ajax handshake/poll: %s (%s)" uid sch-uuid)
-                        updated-conn (upd-conn! :ajax uid client-id server-ch)
+                        updated-conn (upd-conn! :ajax uid client-id :any server-ch)
                         udt-open     (:udt updated-conn)
                         handshake?   (or (:init? updated-conn) (:handshake? params))]
 
@@ -814,7 +821,7 @@
                           (if websocket? "WebSocket" "Ajax")
                           uid sch-uuid)
 
-                      updated-conn (upd-conn! conn-type uid client-id nil)
+                      updated-conn (upd-conn! conn-type uid client-id server-ch nil)
                       udt-close    (:udt updated-conn)]
 
                   ;; Allow some time for possible reconnects (repoll,