From fdfadd359fd680fefcfcf7bb0790840e63156b07 Mon Sep 17 00:00:00 2001 From: "alonso.torres" Date: Fri, 16 Oct 2026 16:01:43 +0200 Subject: [PATCH] :bug: Apply remote changes under buffered local edits too A remote commit undoes the pending local commits, applies the remote changes and replays the local ones, so the local file ends like the server will: remote first, local last. Only queued commits took part; the ones still waiting in the persistence buffer (up to ~3s of edits) stayed under the remote change. When both touched the same attribute, the remote value won locally but lost on the server. The buffer now lives in the store ([:persistence :buffer]) instead of an rx buffer, so the rebase can reach it. A flush merges it per file into queue entries that keep the list of commits they carry, and the leave-page guard reads buffer and queue from the store. A buffer left by a previous session is queued when persistence starts. --- .../workspace-state-persistence-subtleties.md | 8 +- frontend/src/app/main/data/changes.cljs | 9 +- frontend/src/app/main/data/persistence.cljs | 118 ++++++++++++------ .../frontend_tests/data/persistence_test.cljs | 92 ++++++++++++++ 4 files changed, 183 insertions(+), 44 deletions(-) diff --git a/.serena/memories/frontend/workspace-state-persistence-subtleties.md b/.serena/memories/frontend/workspace-state-persistence-subtleties.md index eee6de48b8..9dbac6ecc2 100644 --- a/.serena/memories/frontend/workspace-state-persistence-subtleties.md +++ b/.serena/memories/frontend/workspace-state-persistence-subtleties.md @@ -18,9 +18,9 @@ ## Commits, undo, persistence - `commit-changes` refuses to create commits unless `:permissions :can-edit` is true. It captures file revn/vern, selected-before, features, tags, undo group, and translation flag into a `::commit` event. A refusal answers with an empty stream, never nil, so callers subscribe uniformly instead of branching on nil. -- Applying a remote commit first rolls back pending local commits, applies the remote changes, then replays pending local redo changes. Index updates are emitted for undo, remote redo, and replayed redo paths. +- Applying a remote commit first rolls back pending local commits (queued entries, then `[:persistence :buffer]`), applies the remote changes, then replays pending local redo changes. Index updates are emitted for undo, remote redo, and replayed redo paths. Leaving the buffer out makes a remote change to the same attr win locally but lose on the server. - Local commits are independently consumed by undo, persistence, WASM model updates, thumbnail/library watchers, and text position-data recalculation. -- Persistence buffers local commits: status becomes pending after about 200ms, commits are flushed after about 3s or `::force-persist`, and buffered commits are merged per file before `:update-file`. +- Persistence buffers local commits in the store (`buffer-commit` → `[:persistence :buffer]`): status becomes pending after about 200ms; the 3s debounce or `::force-persist` emits `flush-buffer`, which merges the buffer per file into `append-commit`s, and each append removes its own commits from the buffer (a commit arriving mid-flush stays). `initialize-persistence-state` flushes a leftover buffer. Queue entries keep `:commits` (id, origin, undo-group, redo/undo changes of each merged commit) next to the merged change lists; the merged `:undo-changes` are in forward commit order, so undo per commit through `:commits`. - Persistence sends revn as the max of the commit revn and locally tracked latest revn; remote commits update that revn tracker. - Persistence is skipped in version preview/read-only mode or without edit permission. - A terminal failure halts the queue on purpose: the edits stay queued so `wait-persisted-or-error` keeps rejecting, and `workspace/versions.cljs` relies on that to avoid snapshotting a file whose edits were never saved. `recover-persistence` lifts the halt when a file is opened again or the role changes. A transport failure still inside its retry window (`slow-retry-pending?`) is resent once (`resume-persistence true`), never released; any other halt emits `release-failed-file` (drops every queued commit of the file that failed) then `resume-persistence`. Opening a file also replays the queued edits of its version onto the loaded data (`pending-file-changes` → `restore-pending-changes` in `bundle-fetched`); a replay that throws keeps the server data and the queue. That is what unblocks the editor without a reload; because the queue is shared, dropping the failed file is also what lets other files' pending edits drain. `release-failed-file` runs at recovery time, never at failure time — clearing the error eagerly would make waiters resolve. @@ -31,8 +31,8 @@ - Resume triggers: backoff timer and the browser `online` event (guarded by `exists? js/window`; re-enters the runner only for a live `:retrying` episode). New local edits during `:retrying` only join the queue: the episode keeps its `:run-id`, so `append-commit` does not re-enter, and the live runner (still waiting on the head's `commit-persisted`) sends them after the head. Re-entering on edits would bypass the backoff and spend an attempt per batched commit. - Tests instant-trigger retries by stubbing `rx/timer` (recording delays to assert the schedule); dynamic bindings do not survive `await` continuations, so no dynamic var for delays. - `save-request-timeout-ms` is 7 minutes and wraps the `update-file` request in `rx/timeout`, failing it as a transient `:network` error so the retry path resends instead of blocking the queue forever. It is deliberately longer than `saving-stall-timeout-ms` (5 min) so the stalled report fires before the retry. -- `sync-unload-guard` keeps `:persistence` in `st/ongoing-tasks` while edits are buffered or queued, which is what holds the browser leave-page warning on. It syncs on `::append-commit`, `::discard-commit` and `::release-failed-file` (every queue change). It is the one effect that must stay in `store.cljs`'s `omitset`, and `store.cljs`'s `ongoing-tasks` watcher is guarded by `exists? js/window` because tests run without a window. -- `leave-signals` (document `visibilitychange`→hidden, window `blur`) emits `::force-persist` when edits are buffered, so leaving the tab/window sends the 3 s batch at once; tests mock `dps/leave-signals` BEFORE `initialize-persistence` runs. pagehide/beforeunload are too late for an async save. +- `sync-unload-guard` keeps `:persistence` in `st/ongoing-tasks` while edits are buffered or queued, which is what holds the browser leave-page warning on. It reads `[:persistence :buffer]` and `:queue` and syncs on `::buffer-commit`, `::append-commit`, `::discard-commit` and `::release-failed-file` (every buffer and queue change). It is the one effect that must stay in `store.cljs`'s `omitset`, and `store.cljs`'s `ongoing-tasks` watcher is guarded by `exists? js/window` because tests run without a window. +- `leave-signals` (document `visibilitychange`→hidden, window `blur`) emits `persist-buffered`, which forces a save when `[:persistence :buffer]` is not empty, so leaving the tab/window sends the 3 s batch at once; tests mock `dps/leave-signals` BEFORE `initialize-persistence` runs. pagehide/beforeunload are too late for an async save. - Exports and the `.penpot` download call `force-persist-files-and-wait file-ids timeout` (waits only for those files, ends on `:error` or timeout, never raises) and then `unsaved-changes? state file-ids` to warn; it reads `refs/persistence`, so tests mock it with `(l/derived :persistence store)`. - `persistence-failed` emits a `save-failed` audit event (`:cause-type :code :retrying :queued-commits`) whenever the user is notified (`notify-failure?`, same rule as the toast), including delegated auth/not-found failures. - Undo transactions can stay open only temporarily; timed-out pending transactions are force-committed after about 20s. Undo entries are capped at 50. diff --git a/frontend/src/app/main/data/changes.cljs b/frontend/src/app/main/data/changes.cljs index 0baea3a2ab..67ce24ab46 100644 --- a/frontend/src/app/main/data/changes.cljs +++ b/frontend/src/app/main/data/changes.cljs @@ -69,10 +69,13 @@ (rx/ignore)))))) (defn- get-pending-commits + "Local commits the server has not saved yet, oldest first: the queued + ones, then the ones still buffered." [{:keys [persistence]}] - (->> (:queue persistence) - (map (d/getf (:index persistence))) - (not-empty))) + (-> [] + (into (map (d/getf (:index persistence))) (:queue persistence)) + (into (:buffer persistence)) + (not-empty))) (def ^:private xf:map-page-id ;; Component-only changes have no :page-id; keep drops those nils so diff --git a/frontend/src/app/main/data/persistence.cljs b/frontend/src/app/main/data/persistence.cljs index 7f08500144..630845fa0e 100644 --- a/frontend/src/app/main/data/persistence.cljs +++ b/frontend/src/app/main/data/persistence.cljs @@ -30,6 +30,7 @@ (declare ^:private persist-commit) (declare ^:private resume-persistence) (declare ^:private slow-retry-cycle) +(declare ^:private flush-buffer) (log/set-level! :warn) @@ -293,6 +294,20 @@ :slow-failures :attempts :retry-token :retry-for))))))) +(defn- without-commits + "Removes from `buffer` the commits listed in `commits`." + [buffer commits] + (let [ids (into #{} (map :id) commits)] + (into [] (remove #(contains? ids (:id %))) buffer))) + +(defn- buffer-commit + "Keeps a local commit in the store until the next flush queues it." + [commit] + (ptk/reify ::buffer-commit + ptk/UpdateEvent + (update [_ state] + (update-in state [:persistence :buffer] (fnil conj []) commit)))) + (defn- append-commit "Event used internally to append the current change to the persistence queue." @@ -307,6 +322,7 @@ (-> pstate (cond-> (not= :error (:status pstate)) (update :run-id d/nilv run-id)) + (update :buffer without-commits (:commits commit)) (update :queue queue-conj id) (update :index assoc id commit))))) @@ -730,20 +746,27 @@ (ptk/reify ::initialize-persistence-state ptk/WatchEvent (watch [_ state _] - (let [{:keys [queue status run-id]} (:persistence state)] - (cond - ;; A commit left queued by a previous session, with no runner. - (queue-needs-a-runner? (:persistence state)) - (rx/of (resume-persistence)) + (let [{:keys [queue status run-id buffer]} (:persistence state)] + (rx/concat + (cond + ;; A commit left queued by a previous session, with no runner. + (queue-needs-a-runner? (:persistence state)) + (rx/of (resume-persistence)) - ;; Nothing queued and nothing in flight: there is nothing to save. - (and (empty? queue) - (not= status :error) - (or run-id (#{:pending :saving} status))) - (rx/of (update-status :saved)) + ;; Nothing queued and nothing in flight: there is nothing to save. + (and (empty? queue) + (empty? buffer) + (not= status :error) + (or run-id (#{:pending :saving} status))) + (rx/of (update-status :saved)) - :else - (rx/empty)))))) + :else + (rx/empty)) + ;; Commits buffered by a previous session are queued now instead of + ;; waiting for the next edit. + (if (seq buffer) + (rx/of (flush-buffer)) + (rx/empty))))))) (defn- recover-persistence "Lifts a halted queue when a file is opened again or the role changes. @@ -802,6 +825,10 @@ (def ^:private xf-mapcat-redo (mapcat :redo-changes)) +(def ^:private commit-keys + "What a queue entry keeps of each commit merged into it." + [:id :origin :undo-group :redo-changes :undo-changes]) + (defn- merge-commit [buffer] (->> (rx/from (group-by :file-id buffer)) @@ -811,7 +838,19 @@ (-> item (assoc :undo-changes uchg) (assoc :redo-changes rchg) - (assoc :changes rchg))))))) + (assoc :changes rchg) + (assoc :commits (mapv #(select-keys % commit-keys) commits)))))))) + +(defn- flush-buffer + "Queues the buffered commits, merged into one queue entry per file." + [] + (ptk/reify ::flush-buffer + ptk/WatchEvent + (watch [_ state _] + (if-let [buffer (seq (dm/get-in state [:persistence :buffer]))] + (->> (merge-commit buffer) + (rx/map append-commit)) + (rx/empty))))) (defn- resume-on-online "Re-enters the runner for a retrying episode when the browser reports @@ -830,14 +869,25 @@ (defn- sync-unload-guard "Turns the browser's leave-page warning on while edits are unsaved: buffered, queued, or retained after a failed save." - [buffered?] + [] (ptk/reify ::sync-unload-guard ptk/EffectEvent (effect [_ state _] - (if (or buffered? (seq (dm/get-in state [:persistence :queue]))) + (if (or (seq (dm/get-in state [:persistence :buffer])) + (seq (dm/get-in state [:persistence :queue]))) (swap! st/ongoing-tasks conj :persistence) (swap! st/ongoing-tasks disj :persistence))))) +(defn- persist-buffered + "Forces a save when local commits wait in the buffer." + [] + (ptk/reify ::persist-buffered + ptk/WatchEvent + (watch [_ state _] + (if (seq (dm/get-in state [:persistence :buffer])) + (rx/of ::force-persist) + (rx/empty))))) + (defn leave-signals "Emits when the tab is hidden or the window loses focus: another tab or app, a plugin iframe, a closed laptop. Empty outside the browser." @@ -857,9 +907,6 @@ (log/debug :hint "initialize persistence") (let [stoper-s (rx/filter (ptk/type? ::initialize-persistence) stream) - ;; True while local commits wait in the buffer for the next flush. - buffered? (volatile! false) - local-commits-s (->> stream (rx/filter dch/commit?) @@ -909,23 +956,20 @@ ;; Sends the buffered edits when the user leaves the tab or window. (->> (leave-signals) - (rx/filter (fn [_] @buffered?)) - (rx/map (fn [_] ::force-persist)) + (rx/map (fn [_] (persist-buffered))) (rx/take-until stoper-s)) - ;; Syncs the leave-page warning on every local commit and on every - ;; queue change (append after a flush, discard after a save, - ;; release of a failed file). + ;; Syncs the leave-page warning on every buffer and queue change + ;; (a buffered commit, an append after a flush, a discard after a + ;; save, the release of a failed file). (->> (rx/merge - (rx/of false) - (->> local-commits-s - (rx/map (fn [_] (vreset! buffered? true)))) + (rx/of nil) (->> stream - (rx/filter #(or (ptk/type? ::append-commit %) + (rx/filter #(or (ptk/type? ::buffer-commit %) + (ptk/type? ::append-commit %) (ptk/type? ::discard-commit %) - (ptk/type? ::release-failed-file %))) - (rx/map (fn [_] @buffered?)))) - (rx/map sync-unload-guard) + (ptk/type? ::release-failed-file %))))) + (rx/map (fn [_] (sync-unload-guard))) (rx/take-until stoper-s)) (->> local-commits-s @@ -934,15 +978,15 @@ (update-status :pending))) (rx/take-until stoper-s)) - ;; Here we watch for local commits, buffer them in a small - ;; chunks (very near in time commits) and append them to the - ;; persistence queue + ;; Local commits wait in the store buffer and are queued in small + ;; chunks (commits very near in time) on each notification (->> local-commits-s + (rx/map buffer-commit) + (rx/take-until stoper-s)) + + (->> notifier-s + (rx/map (fn [_] (flush-buffer))) (rx/take-until stoper-s) - (rx/buffer-until notifier-s) - (rx/tap (fn [_] (vreset! buffered? false))) - (rx/mapcat merge-commit) - (rx/map append-commit) (rx/finalize (fn [] (log/debug :hint "finalize persistence: changes watcher")))) diff --git a/frontend/test/frontend_tests/data/persistence_test.cljs b/frontend/test/frontend_tests/data/persistence_test.cljs index de85818c38..f7908d1d25 100644 --- a/frontend/test/frontend_tests/data/persistence_test.cljs +++ b/frontend/test/frontend_tests/data/persistence_test.cljs @@ -29,6 +29,7 @@ [app.main.refs :as refs] [app.main.repo :as rp] [app.main.store :as st] + [app.main.worker :as mw] [app.util.i18n :as i18n] [beicon.v2.core :as rx] [cljs.test :as t :include-macros true] @@ -1136,3 +1137,94 @@ [:cause-type :code :retrying :queued-commits]))))) (fn [_ _] (rx/throw (ex-info "invalid" {:type :validation :code :referential-integrity})))))) + +;; Buffered commits. + +;; Scenario: an edit waits in the buffer for the next flush, then a flush +;; moves it to the queue. Proves: buffered edits are visible in the store +;; until they are queued, so a rollback can reach them. +(t/deftest ^:async buffered-commits-stay-in-the-store-until-queued + (await + (with-persistence + (^:async fn [{:keys [file-id response requests store]}] + (let [commit (local-commit file-id)] + (ptk/emit! store commit) + (await (async/wait-for #(seq (get-in @store [:persistence :buffer])) + "the edit is buffered")) + (t/is (= [(:id @commit)] (mapv :id (get-in @store [:persistence :buffer])))) + (t/is (empty? (get-in @store [:persistence :queue]))) + (ptk/emit! store ::dps/force-persist) + (await (async/wait-for #(seq @requests) "the flush sends the edit")) + (t/is (empty? (get-in @store [:persistence :buffer]))) + (t/is (= [(:id @commit)] + (->> (get-in @store [:persistence :queue]) + (mapcat #(get-in @store [:persistence :index % :commits])) + (mapv :id)))) + (rx/push! response {:revn 1}) + (await (async/wait-for #(= :saved (get-in @store [:persistence :status])) + "the save lands"))))))) + +(defn- rename + [page-id shape-id name] + {:type :mod-obj :id shape-id :page-id page-id + :operations [{:type :set :attr :name :val name}]}) + +(defn- rename-commit + "A local commit of `file-id` that renames the shape from `from` to `to`, + with its own undo group." + [file-id page-id shape-id from to] + (ptk/data-event ::dch/commit + {:id (uuid/next) + :file-id file-id + :file-revn 0 + :file-vern 0 + :source :local + :features #{} + :undo-group (uuid/next) + :redo-changes [(rename page-id shape-id to)] + :undo-changes [(rename page-id shape-id from)]})) + +(defn- file-data + "File data with one page holding one rectangle named `name`." + [page-id shape-id name] + {:pages [page-id] + :pages-index {page-id {:id page-id + :name "Page" + :objects {uuid/zero (cts/setup-shape {:id uuid/zero :type :frame :shapes [shape-id]}) + shape-id (cts/setup-shape {:id shape-id :type :rect :name name + :parent-id uuid/zero :frame-id uuid/zero})}}}}) + +(defn- shape-name + [store file-id page-id shape-id] + (get-in @store [:files file-id :data :pages-index page-id :objects shape-id :name])) + +;; Scenario: an edit waits in the buffer when another user's change to the +;; same attribute arrives. The server will apply the remote change first and +;; the buffered edit after it, so the local file must end the same way. +;; Proves: a remote commit is applied under buffered edits too, not only +;; under queued ones. +(t/deftest ^:async a-remote-commit-goes-under-buffered-edits + (await + (with-persistence + (^:async fn [{:keys [file-id store]}] + (await + (mock/with-mocks* + {mw/ask! (mock/stub (fn [_] (rx/empty)))} + (let [page-id (uuid/next) + shape-id (uuid/next) + local (rename-commit file-id page-id shape-id "Original" "Local")] + (ptk/emit! store #(-> % + (assoc :current-page-id page-id) + (assoc-in [:files file-id :data] (file-data page-id shape-id "Local")))) + (ptk/emit! store local) + (await (async/wait-for #(seq (get-in @store [:persistence :buffer])) + "the local edit is buffered")) + (ptk/emit! store (dch/commit {:file-id file-id + :file-revn 1 + :file-vern 0 + :source :remote + :save-undo? false + :redo-changes [(rename page-id shape-id "Remote")] + :undo-changes []})) + (await (async/settle)) + (t/is (= "Local" (shape-name store file-id page-id shape-id))))))))))