🐛 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.
This commit is contained in:
alonso.torres 2026-10-16 16:01:43 +02:00
parent a6c225d510
commit fdfadd359f
4 changed files with 183 additions and 44 deletions

View File

@ -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.

View File

@ -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

View File

@ -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"))))

View File

@ -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))))))))))