diff --git a/CHANGES.md b/CHANGES.md index 7b7b454..47497ac 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -20,6 +20,15 @@ ### Enhancements +* The client now reconnects automatically when the connection drops (page + reload, server restart, flaky network) using an exponential backoff. It is + on by default and tunable via the `:reconnect?`, `:reconnect-delay` and + `:max-reconnect-delay` options. Call `weasel.repl/disconnect` to close the + connection and stop reconnecting. +* Added an optional application-level heartbeat (`:ping`/`:pong`) that detects a + silently dead connection and triggers a reconnect. It is off by default and + enabled via the `:heartbeat-interval` option; it never disrupts a server that + doesn't answer pings. * Ship a `deps.edn` so the library can be consumed via the Clojure CLI / tools.deps. * Add a GitHub Actions CI pipeline and a basic test suite, including a Node round-trip integration test that exercises the full eval cycle over a real diff --git a/README.md b/README.md index c0889a9..7875f75 100644 --- a/README.md +++ b/README.md @@ -77,9 +77,27 @@ You may optionally specify the following: ; or any variadic function to handle printing differently. ; defaults to :repl :on-open, :on-error, :on-close ; fns for handling websocket lifecycle events. - ; default for all is nil + ; default for all is nil. :on-open and :on-close + ; fire once per connection (not on every + ; reconnect); :on-error receives the native + ; WebSocket event. +:reconnect? ; boolean, whether to reconnect automatically when the + ; connection drops. defaults to true +:reconnect-delay ; initial reconnect backoff in ms, defaults to 1000. + ; the delay doubles after each failed attempt +:max-reconnect-delay ; ceiling for the backoff in ms, defaults to 30000 +:heartbeat-interval ; ms between keepalive pings used to detect a silently + ; dead connection. 0 disables it, which is the default. + ; a few missed pongs are tolerated before the link is + ; considered dead, and a server that never answers pings + ; is left undisturbed ``` +The connection survives page reloads and server restarts on its own, retrying +with the backoff until it gets back in (or the server rejects it outright, e.g. +when another client already holds the REPL). Call `(repl/disconnect)` to close +it for good and stop the reconnection attempts. + Connecting with options: ```clojure (repl/connect "ws://localhost:9001" diff --git a/dev/weasel/integration.clj b/dev/weasel/integration.clj index 3b985e8..b4d8509 100644 --- a/dev/weasel/integration.clj +++ b/dev/weasel/integration.clj @@ -1,10 +1,13 @@ (ns weasel.integration - "End-to-end check of the native-WebSocket transport. + "End-to-end checks of the native-WebSocket transport against a real server + driving the compiled Node client. Exercises, in one run: - Starts a real Weasel server, launches the compiled ClojureScript client - under Node and drives a single round-trip that exercises both the result - and the print paths: the evaluated snippet prints a string and returns a - value, and we assert both travel back over the socket. Run with: + * a round-trip that both prints and returns a value + * the heartbeat (the client keeps pinging, the server pongs, and the + connection is not torn down) + * auto-reconnect after the server is bounced + + Run with: lein cljsbuild once node lein run -m weasel.integration" @@ -12,48 +15,88 @@ [weasel.repl.server :as server])) (def ^:private client-js "target/node/weasel_node_client.js") +(def ^:private port 9009) +(def ^:private heartbeat-ms 300) + +;; the current round's signal promises, swapped out between rounds +(def ^:private signals (atom nil)) +(def ^:private ping-count (atom 0)) +(def ^:private ready-count (atom 0)) + +(defn- fresh-round! [] + (reset! signals {:ready (promise) :printed (promise) :result (promise)})) -(defn- await! - "Derefs promise `p`, throwing if it does not resolve within 10 seconds." - [what p] +(defn- handle [data] + (let [msg (edn/read-string data) + {:keys [ready printed result]} @signals] + (case (:op msg) + :ready (do (swap! ready-count inc) (deliver ready true)) + :print (deliver printed (:value msg)) + :result (deliver result (:value msg)) + :ping (do (swap! ping-count inc) + (server/send! (pr-str {:op :pong}))) + nil))) + +(defn- await! [what p] (let [v (deref p 10000 ::timeout)] (when (= v ::timeout) (throw (ex-info (str "timed out waiting for " what) {}))) v)) +(defn- wait-for-client! [what] + ;; like server/wait-for-client, but bounded so a broken reconnect fails the + ;; run instead of hanging forever + (when (= ::timeout (deref (server/channel) 10000 ::timeout)) + (throw (ex-info (str "timed out waiting for " what) {})))) + +(defn- eval! [code] + (server/send! (pr-str {:op :eval-js :code code}))) + (defn -main [& _] - (let [ready (promise) - printed (promise) - result (promise) - port 9009 - ok? (atom false)] - (server/start - (fn [data] - (let [msg (edn/read-string data)] - (case (:op msg) - :ready (deliver ready true) - :print (deliver printed (:value msg)) - :result (deliver result (:value msg)) - nil))) - :ip "127.0.0.1" :port port) - (let [^"[Ljava.lang.String;" cmd (into-array String ["node" client-js (str "ws://127.0.0.1:" port)]) + (let [ok? (atom false)] + (fresh-round!) + (server/start handle :ip "127.0.0.1" :port port) + (let [^"[Ljava.lang.String;" cmd (into-array String + ["node" client-js (str "ws://127.0.0.1:" port) + (str heartbeat-ms)]) proc (-> (ProcessBuilder. cmd) (.inheritIO) (.start))] (try - (server/wait-for-client) - (await! ":ready" ready) - (server/send! - (pr-str {:op :eval-js - :code "(function () { weasel.repl.repl_print('hi from node'); return 40 + 2; })()"})) - (let [value (:value (await! ":result" result)) - out (await! ":print" printed)] - (println "RESULT:" (pr-str value) " PRINT:" (pr-str out)) - (reset! ok? (and (= value "42") - (= out (pr-str "hi from node"))))) - (catch Exception e + ;; round 1 - result and print travel back over the socket + (wait-for-client! "client to connect") + (await! ":ready" (:ready @signals)) + (eval! "(function () { weasel.repl.repl_print('hi from node'); return 40 + 2; })()") + (let [value (:value (await! ":result" (:result @signals))) + out (await! ":print" (:printed @signals))] + (assert (= value "42") (str "eval result was " (pr-str value))) + (assert (= out (pr-str "hi from node")) (str "print was " (pr-str out)))) + + ;; heartbeat - over several intervals the client keeps pinging (proving + ;; it processes pongs and the cycle continues) and the healthy socket is + ;; NOT torn down (no spurious reconnect, so still exactly one :ready) + (Thread/sleep (long (* 5 heartbeat-ms))) + (assert (>= @ping-count 2) + (str "expected repeated heartbeat pings, saw " @ping-count)) + (assert (= 1 @ready-count) + (str "heartbeat tore down a healthy socket; :ready count = " @ready-count)) + + ;; round 2 - bounce the server, the client reconnects on its own + (fresh-round!) + (server/stop) + (Thread/sleep 200) + (server/start handle :ip "127.0.0.1" :port port) + (wait-for-client! "client to reconnect") + (await! ":ready (after reconnect)" (:ready @signals)) + (eval! "7 * 6") + (assert (= "42" (:value (await! ":result (after reconnect)" (:result @signals)))) + "reconnected eval returned the wrong value") + + (reset! ok? true) + (println (str "PASS - eval, print, heartbeat (" @ping-count + " pings) and reconnect all verified")) + (catch Throwable e (println "FAIL:" (.getMessage e))) (finally (.destroy proc) (server/stop)))) - (if @ok? - (println "PASS") - (do (println "FAIL") (System/exit 1))))) + (when-not @ok? + (System/exit 1)))) diff --git a/src/clj/weasel/repl/websocket.clj b/src/clj/weasel/repl/websocket.clj index c18e617..2fbc02c 100644 --- a/src/clj/weasel/repl/websocket.clj +++ b/src/clj/weasel/repl/websocket.clj @@ -57,6 +57,15 @@ :ready [_ _]) +(defmethod process-message + :ping + [_ _] + (server/send! (pr-str {:op :pong}))) + +(defmethod process-message + :default + [_ _]) + (defn- websocket-setup-env [this opts] (reset! repl-out *out*) diff --git a/src/cljs/weasel/repl.cljs b/src/cljs/weasel/repl.cljs index c33fe87..9af856d 100644 --- a/src/cljs/weasel/repl.cljs +++ b/src/cljs/weasel/repl.cljs @@ -3,13 +3,52 @@ [cljs.reader :as reader :refer [read-string]] [weasel.impls.websocket :as ws])) -(def ^:private ws-connection (atom nil)) +;; Holds the state of the (single) REPL connection, or nil when disconnected. +;; +;; {:id generation token; sockets from a superseded connect are +;; ignored by comparing against it +;; :url the server url, reused when reconnecting +;; :options the normalized connect options +;; :socket the current WebSocket, or nil between attempts +;; :fatal? true once the server rejected us (e.g. :occupied) +;; :opened? true once the user :on-open callback has fired +;; :backoff the current reconnect delay, in ms +;; :reconnect-timer the pending reconnect timeout id, or nil +;; :heartbeat-timer the heartbeat interval id, or nil +;; :stable-timer a one-shot timer that, if the socket survives, clears the +;; backoff, or nil +;; :awaiting-pongs pings sent since the last pong (0 while healthy) +;; :pong-seen? true once the current socket has answered a ping} +(def ^:private connection (atom nil)) + +;; Monotonic source of generation tokens, one per connect call. +(def ^:private generation (atom 0)) + +;; How many pings a proven peer may leave unanswered before we call it dead and +;; tear the socket down. Tolerating more than one ride out network jitter, GC +;; pauses and a momentarily busy server. +(def ^:private max-missed-pongs 2) + +;; How many unanswered pings to send a peer that has never ponged before giving +;; up on the heartbeat (without disturbing the connection itself). A merely slow +;; first pong, arriving within this window, keeps the heartbeat alive. +(def ^:private max-unanswered-pings 3) + +(declare open! schedule-reconnect!) (defn alive? "Returns truthy value if the REPL is attempting to connect or is connected, or falsy value otherwise." [] - (some? @ws-connection)) + (some? @connection)) + +(defn- current? + "Is `id` the generation of the live connection?" + [id] + (= id (:id @connection))) + +(defn- options [] + (:options @connection)) (defn- browser? "Returns true when running in a browser-like environment that can load @@ -22,7 +61,15 @@ (defmethod process-message :error [message] - (.error js/console (str "Websocket REPL error " (:type message)))) + (.error js/console (str "Websocket REPL error " (:type message))) + nil) + +(defmethod process-message + :pong + [_] + ;; clear the outstanding pings without resurrecting a disconnected atom + (swap! connection (fn [c] (when c (assoc c :awaiting-pongs 0 :pong-seen? true)))) + nil) (defmethod process-message :eval-js @@ -44,8 +91,8 @@ (defn repl-print [& args] - (when-let [conn @ws-connection] - (ws/send! conn (pr-str {:op :print :value (apply pr-str args)})))) + (when-let [socket (:socket @connection)] + (ws/send! socket (pr-str {:op :print :value (apply pr-str args)})))) (defn console-print [& args] (.apply (.-log js/console) js/console (into-array args))) @@ -57,39 +104,183 @@ (apply console-print args) (apply repl-print args))}) +(defn- clear-socket-timers! + "Cancels the per-socket heartbeat and stability timers." + [] + (when-let [{:keys [heartbeat-timer stable-timer]} @connection] + (when heartbeat-timer (js/clearInterval heartbeat-timer)) + (when stable-timer (js/clearTimeout stable-timer)) + (swap! connection assoc :heartbeat-timer nil :stable-timer nil))) + +(defn- start-heartbeat! + "Pings the server every `:heartbeat-interval` ms. A proven peer (one that has + answered a ping on this socket) that then misses `max-missed-pongs` pings is + treated as dead and the socket is torn down, triggering a reconnect. A peer + that never answers is simply left alone after `max-unanswered-pings`, so the + heartbeat never disrupts a server that doesn't speak it." + [id socket] + (let [interval (:heartbeat-interval (options))] + (when (and interval (pos? interval)) + (let [timer (js/setInterval + (fn [] + (when (current? id) + (let [{:keys [awaiting-pongs pong-seen?]} @connection] + (cond + (and pong-seen? (>= awaiting-pongs max-missed-pongs)) + (ws/close! socket) + + (and (not pong-seen?) (>= awaiting-pongs max-unanswered-pings)) + (clear-socket-timers!) + + :else + (do + (swap! connection update :awaiting-pongs inc) + (ws/send! socket (pr-str {:op :ping}))))))) + interval)] + (swap! connection assoc :heartbeat-timer timer))))) + +(defn- mark-stable! + "Once a socket has stayed open for the base reconnect delay we trust it and + reset the backoff, so only a genuinely flapping server keeps backing off." + [id socket] + (let [delay (:reconnect-delay (options)) + timer (js/setTimeout + (fn [] + (when (and (current? id) (identical? socket (:socket @connection))) + (swap! connection assoc :backoff delay))) + delay)] + (swap! connection assoc :stable-timer timer))) + +(defn- finish! + "Permanently tears the connection down and fires the user :on-close once." + [] + (let [on-close (:on-close (options))] + (reset! connection nil) + (when (fn? on-close) (on-close)))) + +(defn- make-handlers [id] + {:on-open + (fn [socket] + (when (current? id) + (let [{:keys [verbose print on-open]} (options)] + (swap! connection assoc + :socket socket + :awaiting-pongs 0 + :pong-seen? false + :reconnect-timer nil) + (set-print-fn! (if (fn? print) print (get print-fns print))) + (ws/send! socket (pr-str {:op :ready})) + (start-heartbeat! id socket) + (mark-stable! id socket) + (when verbose (.info js/console "Opened Websocket REPL connection")) + ;; the user :on-open callback is one-shot, not re-run on every reconnect + (when-not (:opened? @connection) + (swap! connection assoc :opened? true) + (when (fn? on-open) (on-open)))))) + + :on-message + (fn [socket data] + (when (current? id) + (let [message (read-string data)] + (when (and (= :error (:op message)) (= :occupied (:type message))) + (swap! connection assoc :fatal? true)) + (when-let [response (process-message message)] + (ws/send! socket (pr-str response)))))) + + :on-close + (fn [_] + (when (current? id) + (clear-socket-timers!) + (swap! connection assoc :socket nil) + (when (:verbose (options)) (.info js/console "Closed Websocket REPL connection")) + (if (and (:reconnect? (options)) (not (:fatal? @connection))) + (schedule-reconnect! id) + (finish!)))) + + :on-error + (fn [evt] + (when (current? id) + (let [{:keys [verbose on-error]} (options)] + (when verbose (.error js/console "WebSocket error" evt)) + (when (fn? on-error) (on-error evt)))))}) + +(defn- open! + "Opens a fresh socket to the configured url for generation `id`. On failure + (e.g. a runtime with no WebSocket) the connection is torn down so `alive?` + reflects reality; the caller decides whether to surface the error." + [id] + (when (current? id) + (try + (let [socket (ws/connect! (:url @connection) (make-handlers id))] + (swap! connection assoc :socket socket :reconnect-timer nil) + socket) + (catch :default e + (reset! connection nil) + (throw e))))) + +(defn- schedule-reconnect! + "Schedules a reconnect after the current backoff, then doubles it up to the + configured ceiling." + [id] + (let [{:keys [max-reconnect-delay verbose]} (options) + delay (:backoff @connection)] + (when verbose + (.info js/console (str "Reconnecting Weasel REPL in " delay "ms"))) + ;; swallow a synchronous open! failure here so a transient error doesn't + ;; kill the reconnect loop with an uncaught exception in the timer + (let [timer (js/setTimeout (fn [] (try (open! id) (catch :default _ nil))) delay)] + (swap! connection assoc + :reconnect-timer timer + :backoff (min max-reconnect-delay (* 2 delay)))))) + +(defn disconnect + "Closes the REPL connection and cancels any pending reconnect attempts." + [] + (when-let [{:keys [socket reconnect-timer]} @connection] + (clear-socket-timers!) + (when reconnect-timer (js/clearTimeout reconnect-timer)) + ;; tear the state down synchronously so alive? is immediately false; the + ;; socket's eventual onclose no-ops because its generation is gone + (finish!) + (when socket (ws/close! socket)))) + (defn connect - [repl-server-url & {:keys [verbose on-open on-error on-close print] - :or {verbose true, print :repl}}] - (let [repl-connection - (ws/connect! repl-server-url - {:on-open - (fn [socket] - (set-print-fn! (if (fn? print) print (get print-fns print))) - (ws/send! socket (pr-str {:op :ready})) - (when verbose (.info js/console "Opened Websocket REPL connection")) - (when (fn? on-open) (on-open))) - - :on-message - (fn [socket data] - (let [message (read-string data) - response (-> message process-message pr-str)] - (ws/send! socket response))) - - :on-close - (fn [_] - (reset! ws-connection nil) - (when verbose (.info js/console "Closed Websocket REPL connection")) - (when (fn? on-close) (on-close))) - - :on-error - (fn [evt] - (when verbose (.error js/console "WebSocket error" evt)) - (when (fn? on-error) (on-error evt)))})] - - (reset! ws-connection repl-connection) + [repl-server-url & {:keys [verbose on-open on-error on-close print + reconnect? reconnect-delay max-reconnect-delay + heartbeat-interval] + :or {verbose true + print :repl + reconnect? true + reconnect-delay 1000 + max-reconnect-delay 30000 + heartbeat-interval 0}}] + ;; replace any existing connection cleanly rather than leaking its timers + (when @connection (disconnect)) + (let [id (swap! generation inc) + options {:verbose verbose + :on-open on-open + :on-error on-error + :on-close on-close + :print print + :reconnect? reconnect? + :reconnect-delay reconnect-delay + :max-reconnect-delay max-reconnect-delay + :heartbeat-interval heartbeat-interval}] + (reset! connection {:id id + :url repl-server-url + :options options + :socket nil + :fatal? false + :opened? false + :backoff reconnect-delay + :reconnect-timer nil + :heartbeat-timer nil + :stable-timer nil + :awaiting-pongs 0 + :pong-seen? false}) ;; reusable bootstrap - only meaningful in a browser, where new code is ;; loaded by appending script tags to the document (when (browser?) (brepl/bootstrap)) - repl-connection)) + (open! id))) diff --git a/test/clj/weasel/repl/websocket_test.clj b/test/clj/weasel/repl/websocket_test.clj new file mode 100644 index 0000000..4619a36 --- /dev/null +++ b/test/clj/weasel/repl/websocket_test.clj @@ -0,0 +1,12 @@ +(ns weasel.repl.websocket-test + (:require [clojure.test :refer [deftest is testing]] + [clojure.edn :as edn] + [weasel.repl.server :as server] + [weasel.repl.websocket :as websocket])) + +(deftest ping-is-answered-with-pong + (testing "a :ping message makes the server send a :pong back" + (let [sent (atom nil)] + (with-redefs [server/send! (fn [msg] (reset! sent msg))] + (#'websocket/process-message ::ignored {:op :ping})) + (is (= {:op :pong} (edn/read-string @sent)))))) diff --git a/test/cljs/weasel/node_client.cljs b/test/cljs/weasel/node_client.cljs index 38c4652..1d7cbb6 100644 --- a/test/cljs/weasel/node_client.cljs +++ b/test/cljs/weasel/node_client.cljs @@ -1,10 +1,15 @@ (ns weasel.node-client "A tiny Node entry point used by the integration test to prove that the - REPL client works outside the browser, on a native `WebSocket`." + REPL client works outside the browser, on a native `WebSocket`. + + Args: [heartbeat-interval-ms]" (:require [weasel.repl :as repl])) (defn -main [& args] - (let [url (or (first args) "ws://127.0.0.1:9001")] - (repl/connect url :verbose false))) + (let [url (or (first args) "ws://127.0.0.1:9001") + hb (some-> (second args) (js/parseInt 10))] + (apply repl/connect url + :verbose false + (if hb [:heartbeat-interval hb] [])))) (set! *main-cli-fn* -main)