From a261e00a3337f95b7887a7a5ab6373b458f6fe5b Mon Sep 17 00:00:00 2001 From: "alonso.torres" Date: Wed, 4 Nov 2026 09:57:56 +0100 Subject: [PATCH] :bug: Keep a file reload from doubling the workspace streams Reloading the open file (a peer's restore, the fallback of a rollback) runs initialize-workspace again without leaving the workspace. The streams the first run started kept running next to the new ones, so a commit added one undo entry per reload and each notification was handled twice. The old subscription also sent its unsubscribe after the new subscribe, which ended the new one on the server. A new initialize-workspace, text reflow, shape layout or notification subscription stops the one before it, and a subscription replaced by one to the same file sends no unsubscribe. --- frontend/src/app/main/data/workspace.cljs | 5 +- .../main/data/workspace/notifications.cljs | 27 +++++- .../app/main/data/workspace/shape_layout.cljs | 5 +- .../src/app/main/data/workspace/texts.cljs | 5 +- .../data/workspace_notifications_test.cljs | 94 +++++++++++++++++++ .../data/workspace_undo_test.cljs | 36 +++++++ 6 files changed, 167 insertions(+), 5 deletions(-) diff --git a/frontend/src/app/main/data/workspace.cljs b/frontend/src/app/main/data/workspace.cljs index a42c367c3b..28e14bc3bd 100644 --- a/frontend/src/app/main/data/workspace.cljs +++ b/frontend/src/app/main/data/workspace.cljs @@ -416,7 +416,10 @@ ptk/WatchEvent (watch [_ state stream] - (let [stoper-s (rx/filter (ptk/type? ::finalize-workspace) stream) + (let [;; A reload starts a new `initialize-workspace`, which replaces these streams + stoper-s (rx/filter #(or (ptk/type? ::finalize-workspace %) + (ptk/type? ::initialize-workspace %)) + stream) rparams (rt/get-params state) features (features/get-enabled-features state team-id) render-wasm-enabled? (features/active-feature? state "render-wasm/v1") diff --git a/frontend/src/app/main/data/workspace/notifications.cljs b/frontend/src/app/main/data/workspace/notifications.cljs index b2637bbce7..2cd1562589 100644 --- a/frontend/src/app/main/data/workspace/notifications.cljs +++ b/frontend/src/app/main/data/workspace/notifications.cljs @@ -48,12 +48,29 @@ (declare handle-pointer-send) (declare handle-export-update) +(defn- resubscribed? + "True when `event` is a new subscription to `file-id`, which a reload of + the open file starts." + [file-id event] + (and (ptk/type? ::initialize event) + (= file-id (:file-id (deref event))))) + (defn initialize + "Subscribes to the notifications of `file-id` and its team until the file + is left or a new subscription starts. A new subscription to the same file + replaces this one, so a reload never doubles the handled messages." [team-id file-id] (ptk/reify ::initialize + IDeref + (-deref [_] {:file-id file-id}) + ptk/WatchEvent (watch [_ state stream] - (let [stopper (rx/filter (ptk/type? ::finalize) stream) + (let [stopped-by (volatile! nil) + stopper (->> stream + (rx/filter #(or (ptk/type? ::finalize %) + (ptk/type? ::initialize %))) + (rx/tap #(vreset! stopped-by %))) profile-id (:profile-id state) organization-id (dm/get-in state [:teams team-id :organization :id]) @@ -107,7 +124,13 @@ (rx/take-until stopper))] - (rx/concat stream (rx/of (dws/send endmsg))))))) + (rx/concat + stream + ;; The server keeps one file subscription per connection: an + ;; unsubscribe sent after the new subscribe would end it + (rx/defer #(if (resubscribed? file-id @stopped-by) + (rx/empty) + (rx/of (dws/send endmsg))))))))) (defn- handle-change-team-role [{:keys [role] :as msg}] diff --git a/frontend/src/app/main/data/workspace/shape_layout.cljs b/frontend/src/app/main/data/workspace/shape_layout.cljs index df614dfe7a..9062ad1c5c 100644 --- a/frontend/src/app/main/data/workspace/shape_layout.cljs +++ b/frontend/src/app/main/data/workspace/shape_layout.cljs @@ -150,7 +150,10 @@ (ptk/reify ::initialize-shape-layout ptk/WatchEvent (watch [_ _ stream] - (let [stopper (rx/filter (ptk/type? ::finalize-shape-layout) stream)] + (let [;; A workspace reload starts a new one, which replaces this one + stopper (rx/filter #(or (ptk/type? ::finalize-shape-layout %) + (ptk/type? ::initialize-shape-layout %)) + stream)] (->> stream (rx/filter (ptk/type? :layout/update)) ;; Keeps each update's page and the shapes that can be laid out; diff --git a/frontend/src/app/main/data/workspace/texts.cljs b/frontend/src/app/main/data/workspace/texts.cljs index 4b533c8fa5..0464cee98b 100644 --- a/frontend/src/app/main/data/workspace/texts.cljs +++ b/frontend/src/app/main/data/workspace/texts.cljs @@ -96,7 +96,10 @@ (ptk/reify ::initialize-text-reflow ptk/WatchEvent (watch [_ _ stream] - (let [stopper (rx/filter (ptk/type? ::finalize-text-reflow) stream) + (let [;; A workspace reload starts a new one, which replaces this one + stopper (rx/filter #(or (ptk/type? ::finalize-text-reflow %) + (ptk/type? ::initialize-text-reflow %)) + stream) page-stopper (rx/filter page-finalize? stream)] (->> stream (rx/filter (ptk/type? :text/reflow)) diff --git a/frontend/test/frontend_tests/data/workspace_notifications_test.cljs b/frontend/test/frontend_tests/data/workspace_notifications_test.cljs index 7c6d362941..0bb9d87f95 100644 --- a/frontend/test/frontend_tests/data/workspace_notifications_test.cljs +++ b/frontend/test/frontend_tests/data/workspace_notifications_test.cljs @@ -12,12 +12,14 @@ [app.main.data.changes :as dch] [app.main.data.modal :as modal] [app.main.data.notifications :as ntf] + [app.main.data.websocket :as dws] [app.main.data.workspace :as-alias dw] [app.main.data.workspace.notifications :as dwntf] [app.util.i18n :as i18n] [beicon.v2.core :as rx] [cljs.test :as t :include-macros true] [frontend-tests.helpers.async :as async] + [frontend-tests.helpers.mock :as mock] [potok.v2.core :as ptk])) (defn- file-change-msg @@ -143,3 +145,95 @@ :session-id (uuid/next) :vern 2}))] (t/is (empty? events)))) + +(defn- ^:async subscription-messages + "Runs `scenario` with a store over `team-id`, `dws/send` and the pointer + handler stubbed. `scenario` gets the store and a function that delivers + a peer's pointer message for a file. Resolves to the messages sent over + the websocket and the number of handled pointer messages." + [team-id scenario] + (let [sent (atom []) + handled (atom 0) + result (atom nil) + store (ptk/store {:state {:profile-id (uuid/next) + :session-id (uuid/next) + :teams {team-id {:id team-id}}} + :on-error #(t/is false (str %))}) + pointer (fn [file-id] + (ptk/emit! store (ptk/data-event ::dws/message + {:type :pointer-update + :topic file-id + :file-id file-id + :session-id (uuid/next)})))] + (await + (mock/with-mocks* + {dws/send (mock/stub (fn [msg] + (swap! sent conj msg) + (ptk/data-event ::sent msg))) + dwntf/handle-pointer-update (mock/stub (fn [_] + (swap! handled inc) + (ptk/data-event ::handled {})))} + (try + (await (scenario store pointer)) + ;; Taken before the store ends, which ends the subscriptions too + (reset! result {:sent (mapv #(select-keys % [:type :file-id]) @sent) + :handled @handled}) + (finally + (rx/dispose! store))))) + @result)) + +;; Scenario: the open file reloads three times (a peer's restore, the +;; restorer's own version reload, the rollback fallback), each starting a +;; new subscription, and then a peer's message arrives. Proves: one live +;; subscription per open file, so each message is handled once, and no +;; unsubscribe ends the server's subscription of the reloaded file. +(t/deftest ^:async a-reload-replaces-the-file-subscription + (let [team-id (uuid/next) + file-id (uuid/next) + result (await + (subscription-messages + team-id + (^:async fn [store pointer] + (ptk/emit! store + (dwntf/initialize team-id file-id) + (dwntf/initialize team-id file-id) + (dwntf/initialize team-id file-id)) + (await (async/settle)) + (pointer file-id) + (await (async/settle)))))] + (t/is (= 1 (:handled result)) "the message is handled once") + (t/is (= 3 (count (filter #(= :subscribe-file (:type %)) (:sent result))))) + (t/is (not-any? #(= :unsubscribe-file (:type %)) (:sent result))))) + +;; Scenario: the user leaves a file, and in another case opens another file +;; without leaving the first one. Proves: the subscription ends with an +;; unsubscribe of the file it was for, and the old file's messages are no +;; longer handled. +(t/deftest ^:async leaving-or-switching-files-ends-the-file-subscription + (let [team-id (uuid/next) + file-a (uuid/next) + file-b (uuid/next) + left (await + (subscription-messages + team-id + (^:async fn [store _] + (ptk/emit! store + (dwntf/initialize team-id file-a) + (dwntf/finalize file-a)) + (await (async/settle))))) + switch (await + (subscription-messages + team-id + (^:async fn [store pointer] + (ptk/emit! store + (dwntf/initialize team-id file-a) + (dwntf/initialize team-id file-b)) + (await (async/settle)) + (pointer file-a) + (pointer file-b) + (await (async/settle)))))] + (t/is (= [{:type :unsubscribe-file :file-id file-a}] + (filterv #(= :unsubscribe-file (:type %)) (:sent left)))) + (t/is (= [{:type :unsubscribe-file :file-id file-a}] + (filterv #(= :unsubscribe-file (:type %)) (:sent switch)))) + (t/is (= 1 (:handled switch)) "only the new file's message is handled"))) diff --git a/frontend/test/frontend_tests/data/workspace_undo_test.cljs b/frontend/test/frontend_tests/data/workspace_undo_test.cljs index 465e0b01ba..55a1ae0cca 100644 --- a/frontend/test/frontend_tests/data/workspace_undo_test.cljs +++ b/frontend/test/frontend_tests/data/workspace_undo_test.cljs @@ -12,9 +12,14 @@ [app.common.test-helpers.ids-map :as cthi] [app.common.uuid :as uuid] [app.main.data.changes :as dch] + [app.main.data.comments :as dcmt] + [app.main.data.fonts :as df] + [app.main.data.workspace :as dw] [app.main.data.workspace.undo :as dwu] [beicon.v2.core :as rx] [cljs.test :as t :include-macros true] + [frontend-tests.helpers.async :as async] + [frontend-tests.helpers.mock :as mock] [frontend-tests.helpers.state :as ths] [potok.v2.core :as ptk])) @@ -236,3 +241,34 @@ (t/is (= 2 (:index (forget undo-state #{(uuid/next)} 2)))) (t/is (= 0 (:index (forget undo-state #{g2} 2))) "a cut history cannot point past its end"))) + +;; Scenario: the workspace opens and then reloads twice (a peer's restore, +;; the rollback fallback), each reload running `initialize-workspace` again +;; without leaving the workspace; then one commit is made. Proves: each +;; reload replaces the workspace streams, so the commit adds one undo entry, +;; not one per reload. +(t/deftest ^:async a-workspace-reload-adds-one-undo-entry-per-commit + (let [file (cthf/sample-file :file1 :page-label :page1) + team-id (uuid/next) + store (ths/setup-store file {:renderer :svg}) + loaded (fn [type] (mock/stub (fn [& _] (ptk/data-event type {}))))] + (await + (mock/with-mocks* + {dw/initialize-file (loaded ::file-requested) + dcmt/retrieve-comment-threads (loaded ::threads-requested) + dcmt/fetch-profiles (loaded ::profiles-requested) + df/fetch-fonts (loaded ::fonts-requested)} + (try + (ptk/emit! store + #(assoc % :current-team-id team-id) + (dw/initialize-workspace team-id (:id file)) + (ptk/event ::dw/reload-current-file) + (ptk/event ::dw/reload-current-file)) + (await (async/settle)) + (ptk/emit! store (rename-page (cthf/current-page-id file) "Renamed")) + (await (async/wait-for #(seq (get-in @store [:workspace-undo :items])) + "the commit reaches the undo history")) + (await (async/settle)) + (t/is (= 1 (count (get-in @store [:workspace-undo :items])))) + (finally + (rx/dispose! store)))))))