♻️ Refactor websockets impl to use virtual threads

Removing the use of core.async code and implement code using
plain old and familiar synchronous code

backend/src/app/msgbus.clj

@@ -8,20 +8,18 @@
   "The msgbus abstraction implemented using redis as underlying backend."
   (:require
    [app.common.data :as d]
-   [app.common.exceptions :as ex]
    [app.common.logging :as l]
    [app.common.spec :as us]
    [app.common.transit :as t]
    [app.config :as cfg]
-   [app.redis :as redis]
-   [app.util.async :as aa]
+   [app.redis :as rds]
    [app.util.time :as dt]
    [app.worker :as wrk]
-   [clojure.core.async :as a]
    [clojure.spec.alpha :as s]
    [integrant.core :as ig]
    [promesa.core :as p]
-   [promesa.exec :as px]))
+   [promesa.exec :as px]
+   [promesa.exec.csp :as sp]))
 
 (set! *warn-on-reflection* true)
 
@@ -34,132 +32,116 @@
 (def ^:private xform-prefix-topic
   (map (fn [obj] (update obj :topic prefix-topic))))
 
-(declare ^:private redis-connect)
-(declare ^:private redis-disconnect)
-(declare ^:private redis-pub)
-(declare ^:private redis-sub)
-(declare ^:private redis-unsub)
+(declare ^:private redis-pub!)
+(declare ^:private redis-sub!)
+(declare ^:private redis-unsub!)
 (declare ^:private start-io-loop!)
 (declare ^:private subscribe-to-topics)
 (declare ^:private unsubscribe-channels)
 
-(defmethod ig/prep-key ::msgbus
-  [_ cfg]
-  (merge {:buffer-size 128
-          :timeout (dt/duration {:seconds 30})}
-         (d/without-nils cfg)))
-
-(s/def ::cmd-ch ::aa/channel)
-(s/def ::rcv-ch ::aa/channel)
-(s/def ::pub-ch ::aa/channel)
+(s/def ::cmd-ch sp/chan?)
+(s/def ::rcv-ch sp/chan?)
+(s/def ::pub-ch sp/chan?)
 (s/def ::state ::us/agent)
-(s/def ::pconn ::redis/connection-holder)
-(s/def ::sconn ::redis/connection-holder)
+(s/def ::pconn ::rds/connection-holder)
+(s/def ::sconn ::rds/connection-holder)
 (s/def ::msgbus
   (s/keys :req [::cmd-ch ::rcv-ch ::pub-ch ::state ::pconn ::sconn ::wrk/executor]))
 
-(s/def ::buffer-size ::us/integer)
-
 (defmethod ig/pre-init-spec ::msgbus [_]
-  (s/keys :req-un [::buffer-size ::redis/timeout ::redis/redis ::wrk/executor]))
+  (s/keys :req [::rds/redis ::wrk/executor]))
+
+(defmethod ig/prep-key ::msgbus
+  [_ cfg]
+  (-> cfg
+      (assoc ::buffer-size 128)
+      (assoc ::timeout (dt/duration {:seconds 30}))))
 
 (defmethod ig/init-key ::msgbus
-  [_ {:keys [buffer-size executor] :as cfg}]
+  [_ {:keys [::buffer-size ::wrk/executor ::timeout ::rds/redis] :as cfg}]
   (l/info :hint "initialize msgbus" :buffer-size buffer-size)
-  (let [cmd-ch (a/chan buffer-size)
-        rcv-ch (a/chan (a/dropping-buffer buffer-size))
-        pub-ch (a/chan (a/dropping-buffer buffer-size) xform-prefix-topic)
+  (let [cmd-ch (sp/chan :buf buffer-size)
+        rcv-ch (sp/chan :buf (sp/dropping-buffer buffer-size))
+        pub-ch (sp/chan :buf (sp/dropping-buffer buffer-size)
+                        :xf  xform-prefix-topic)
         state  (agent {})
-        msgbus (-> (redis-connect cfg)
+
+        pconn  (rds/connect redis :timeout timeout)
+        sconn  (rds/connect redis :type :pubsub :timeout timeout)
+        msgbus (-> cfg
+                   (assoc ::pconn pconn)
+                   (assoc ::sconn sconn)
                    (assoc ::cmd-ch cmd-ch)
                    (assoc ::rcv-ch rcv-ch)
                    (assoc ::pub-ch pub-ch)
                    (assoc ::state state)
                    (assoc ::wrk/executor executor))]
 
-    (us/verify! ::msgbus msgbus)
-
     (set-error-handler! state #(l/error :cause % :hint "unexpected error on agent" ::l/sync? true))
     (set-error-mode! state :continue)
-    (start-io-loop! msgbus)
 
-    msgbus))
-
-(defn sub!
-  [{:keys [::state ::wrk/executor] :as cfg} & {:keys [topic topics chan]}]
-  (let [done-ch (a/chan)
-        topics  (into [] (map prefix-topic) (if topic [topic] topics))]
-    (l/debug :hint "subscribe" :topics topics)
-    (send-via executor state subscribe-to-topics cfg topics chan done-ch)
-    done-ch))
-
-(defn pub!
-  [{::keys [pub-ch]} & {:as params}]
-  (a/go
-    (a/>! pub-ch params)))
-
-(defn purge!
-  [{:keys [::state ::wrk/executor] :as msgbus} chans]
-  (l/trace :hint "purge" :chans (count chans))
-  (let [done-ch (a/chan)]
-    (send-via executor state unsubscribe-channels msgbus chans done-ch)
-    done-ch))
+    (assoc msgbus ::io-thr (start-io-loop! msgbus))))
 
 (defmethod ig/halt-key! ::msgbus
   [_ msgbus]
-  (redis-disconnect msgbus)
-  (a/close! (::cmd-ch msgbus))
-  (a/close! (::rcv-ch msgbus))
-  (a/close! (::pub-ch msgbus)))
+  (px/interrupt! (::io-thr msgbus))
+  (sp/close! (::cmd-ch msgbus))
+  (sp/close! (::rcv-ch msgbus))
+  (sp/close! (::pub-ch msgbus))
+  (d/close! (::pconn msgbus))
+  (d/close! (::sconn msgbus)))
+
+(defn sub!
+  [{:keys [::state ::wrk/executor] :as cfg} & {:keys [topic topics chan]}]
+  (let [topics (into [] (map prefix-topic) (if topic [topic] topics))]
+    (l/debug :hint "subscribe" :topics topics :chan (hash chan))
+    (send-via executor state subscribe-to-topics cfg topics chan)
+    nil))
+
+(defn pub!
+  [{::keys [pub-ch]} & {:as params}]
+  (sp/put! pub-ch params))
+
+(defn purge!
+  [{:keys [::state ::wrk/executor] :as msgbus} chans]
+  (l/debug :hint "purge" :chans (count chans))
+  (send-via executor state unsubscribe-channels msgbus chans)
+  nil)
 
 ;; --- IMPL
 
-(defn- redis-connect
-  [{:keys [timeout redis] :as cfg}]
-  (let [pconn (redis/connect redis :timeout timeout)
-        sconn (redis/connect redis :type :pubsub :timeout timeout)]
-    {::pconn pconn
-     ::sconn sconn}))
-
-(defn- redis-disconnect
-  [{:keys [::pconn ::sconn] :as cfg}]
-  (d/close! pconn)
-  (d/close! sconn))
-
 (defn- conj-subscription
   "A low level function that is responsible to create on-demand
   subscriptions on redis. It reuses the same subscription if it is
-  already established. Intended to be executed in agent."
+  already established."
   [nsubs cfg topic chan]
   (let [nsubs (if (nil? nsubs) #{chan} (conj nsubs chan))]
     (when (= 1 (count nsubs))
       (l/trace :hint "open subscription" :topic topic ::l/sync? true)
-      (redis-sub cfg topic))
+      (redis-sub! cfg topic))
     nsubs))
 
 (defn- disj-subscription
   "A low level function responsible on removing subscriptions. The
   subscription is truly removed from redis once no single local
-  subscription is look for it. Intended to be executed in agent."
+  subscription is look for it."
   [nsubs cfg topic chan]
   (let [nsubs (disj nsubs chan)]
     (when (empty? nsubs)
       (l/trace :hint "close subscription" :topic topic ::l/sync? true)
-      (redis-unsub cfg topic))
+      (redis-unsub! cfg topic))
     nsubs))
 
 (defn- subscribe-to-topics
-  "Function responsible to attach local subscription to the
-  state. Intended to be used in agent."
-  [state cfg topics chan done-ch]
-  (aa/with-closing done-ch
-    (let [state (update state :chans assoc chan topics)]
-      (reduce (fn [state topic]
-                (update-in state [:topics topic] conj-subscription cfg topic chan))
-              state
-              topics))))
+  "Function responsible to attach local subscription to the state."
+  [state cfg topics chan]
+  (let [state (update state :chans assoc chan topics)]
+    (reduce (fn [state topic]
+              (update-in state [:topics topic] conj-subscription cfg topic chan))
+            state
+            topics)))
 
-(defn- unsubscribe-single-channel
+(defn- unsubscribe-channel
   "Auxiliary function responsible on removing a single local
   subscription from the state."
   [state cfg chan]
@@ -174,87 +156,113 @@
   "Function responsible from detach from state a seq of channels,
   useful when client disconnects or in-bulk unsubscribe
   operations. Intended to be executed in agent."
-  [state cfg channels done-ch]
-  (aa/with-closing done-ch
-    (reduce #(unsubscribe-single-channel %1 cfg %2) state channels)))
+  [state cfg channels]
+  (reduce #(unsubscribe-channel %1 cfg %2) state channels))
 
 (defn- create-listener
   [rcv-ch]
-  (redis/pubsub-listener
+  (rds/pubsub-listener
    :on-message (fn [_ topic message]
                  ;; There are no back pressure, so we use a slidding
                  ;; buffer for cases when the pubsub broker sends
                  ;; more messages that we can process.
                  (let [val {:topic topic :message (t/decode message)}]
-                   (when-not (a/offer! rcv-ch val)
+                   (when-not (sp/offer! rcv-ch val)
                      (l/warn :msg "dropping message on subscription loop"))))))
 
+(defn- process-input!
+  [{:keys [::state ::wrk/executor] :as cfg} topic message]
+  (let [chans (get-in @state [:topics topic])]
+    (when-let [closed (loop [chans  (seq chans)
+                             closed #{}]
+                        (if-let [ch (first chans)]
+                          (if (sp/put! ch message)
+                            (recur (rest chans) closed)
+                            (recur (rest chans) (conj closed ch)))
+                          (seq closed)))]
+      (send-via executor state unsubscribe-channels cfg closed))))
+
+
 (defn start-io-loop!
   [{:keys [::sconn ::rcv-ch ::pub-ch ::state ::wrk/executor] :as cfg}]
-  (redis/add-listener! sconn (create-listener rcv-ch))
-  (letfn [(send-to-topic [topic message]
-            (a/go-loop [chans  (seq (get-in @state [:topics topic]))
-                        closed #{}]
-              (if-let [ch (first chans)]
-                (if (a/>! ch message)
-                  (recur (rest chans) closed)
-                  (recur (rest chans) (conj closed ch)))
-                (seq closed))))
+  (rds/add-listener! sconn (create-listener rcv-ch))
 
-          (process-incoming [{:keys [topic message]}]
-            (a/go
-              (when-let [closed (a/<! (send-to-topic topic message))]
-                (send-via executor state unsubscribe-channels cfg closed nil))))
-          ]
-    (px/thread
-      {:name "penpot/msgbus-io-loop"}
+  (px/thread
+    {:name "penpot/msgbus/io-loop"
+     :virtual true}
+    (try
       (loop []
-        (let [[val port] (a/alts!! [pub-ch rcv-ch])]
+        (let [timeout-ch (sp/timeout-chan 1000)
+              [val port] (sp/alts! [timeout-ch pub-ch rcv-ch])]
           (cond
-            (nil? val)
-            (do
-              (l/trace :hint "stopping io-loop, nil received")
-              (send-via executor state (fn [state]
-                                         (->> (vals state)
-                                              (mapcat identity)
-                                              (filter some?)
-                                              (run! a/close!))
-                                         nil)))
-
-            (= port rcv-ch)
-            (do
-              (a/<!! (process-incoming val))
+            (identical? port timeout-ch)
+            (let [closed (->> (:chans @state)
+                              (map key)
+                              (filter sp/closed?))]
+              (when (seq closed)
+                (send-via executor state unsubscribe-channels cfg closed)
+                (l/debug :hint "proactively purge channels" :count (count closed)))
               (recur))
 
-            (= port pub-ch)
-            (let [result (a/<!! (redis-pub cfg val))]
-              (when (ex/exception? result)
-                (l/error :hint "unexpected error on publishing"
-                         :message val
-                         :cause result))
-              (recur))))))))
+            (nil? val)
+            (throw (InterruptedException. "internally interrupted"))
 
-(defn- redis-pub
+            (identical? port rcv-ch)
+            (let [{:keys [topic message]} val]
+              (process-input! cfg topic message)
+              (recur))
+
+            (identical? port pub-ch)
+            (do
+              (redis-pub! cfg val)
+              (recur)))))
+
+      (catch InterruptedException _
+        (l/trace :hint "io-loop thread interrumpted"))
+
+      (catch Throwable cause
+        (l/error :hint "unexpected exception on io-loop thread"
+                 :cause cause))
+      (finally
+        (l/trace :hint "clearing io-loop state")
+        (when-let [chans (:chans @state)]
+          (run! sp/close! (keys chans)))
+
+        (l/debug :hint "io-loop thread terminated")))))
+
+
+(defn- redis-pub!
   "Publish a message to the redis server. Asynchronous operation,
   intended to be used in core.async go blocks."
   [{:keys [::pconn] :as cfg} {:keys [topic message]}]
-  (let [message (t/encode message)
-        res     (a/chan 1)]
-    (-> (redis/publish! pconn topic message)
-        (p/finally (fn [_ cause]
-                     (when (and cause (redis/open? pconn))
-                       (a/offer! res cause))
-                     (a/close! res))))
-    res))
+  (try
+    (p/await! (rds/publish! pconn topic (t/encode message)))
+    (catch InterruptedException cause
+      (throw cause))
+    (catch Throwable cause
+      (l/error :hint "unexpected error on publishing"
+               :message message
+               :cause cause))))
 
-(defn redis-sub
+(defn- redis-sub!
   "Create redis subscription. Blocking operation, intended to be used
   inside an agent."
   [{:keys [::sconn] :as cfg} topic]
-  (redis/subscribe! sconn topic))
+  (try
+    (rds/subscribe! sconn topic)
+    (catch InterruptedException cause
+      (throw cause))
+    (catch Throwable cause
+      (l/trace :hint "exception on subscribing" :topic topic :cause cause))))
 
-(defn redis-unsub
+(defn- redis-unsub!
   "Removes redis subscription. Blocking operation, intended to be used
   inside an agent."
   [{:keys [::sconn] :as cfg} topic]
-  (redis/unsubscribe! sconn topic))
+  (try
+    (rds/unsubscribe! sconn topic)
+    (catch InterruptedException cause
+      (throw cause))
+    (catch Throwable cause
+      (l/trace :hint "exception on unsubscribing" :topic topic :cause cause))))
+