From dd163a696de42906bb9edaf57ca7ddf1a6678a8c Mon Sep 17 00:00:00 2001 From: "alonso.torres" Date: Thu, 24 Sep 2026 10:10:02 +0200 Subject: [PATCH] :sparkles: Apply a repeated update-file only once --- backend/src/app/rpc/commands/files_update.clj | 266 +++++++++++++---- backend/test/backend_tests/rpc_file_test.clj | 281 +++++++++++++++++- 2 files changed, 490 insertions(+), 57 deletions(-) diff --git a/backend/src/app/rpc/commands/files_update.clj b/backend/src/app/rpc/commands/files_update.clj index 9d996c6dd4..b159e79013 100644 --- a/backend/src/app/rpc/commands/files_update.clj +++ b/backend/src/app/rpc/commands/files_update.clj @@ -40,6 +40,12 @@ [clojure.set :as set])) (declare ^:private get-lagged-changes) +(declare ^:private commit-key) +(declare ^:private claim-commit!) +(declare ^:private release-commit!) +(declare ^:private record-commit!) +(declare ^:private get-repeated-commit-result) +(declare ^:private update-file-tx) (declare ^:private send-notifications!) (declare ^:private update-file) (declare ^:private update-file*) @@ -47,6 +53,24 @@ (declare ^:private take-snapshot?) (declare ^:private invalidate-caches!) +(def ^:private commit-retention + "How long a commit id is remembered. A client stops retrying a save at this + same age, so a retry never outlives the record that makes it safe to send." + (ct/duration {:hours 24})) + +(def ^:private commit-pending-ttl + "How long a save in progress holds its commit id. Bounds how long a repeat + is turned away when the save never recorded its outcome." + (ct/duration {:minutes 1})) + +(def ^:private commit-await-ms + "How long a repeat waits for the outcome of the save holding its commit id." + 2000) + +(def ^:private commit-poll-ms 50) + +(def ^:private pending-commit "pending") + ;; PUBLIC API; intended to be used outside of this module (declare update-file!) (declare update-file-data!) @@ -62,6 +86,7 @@ [:session-id ::sm/uuid] [:revn {:min 0} ::sm/int] [:vern {:min 0} ::sm/int] + [:commit-id {:optional true} ::sm/uuid] [:features {:optional true} ::cfeat/features] [:changes {:optional true} [:vector cpc/schema:change]] [:changes-with-metadata {:optional true} @@ -140,78 +165,105 @@ ::sm/params schema:update-file ::sm/result schema:update-file-result ::doc/module :files - ::doc/added "1.17" - ::db/transaction true} + ::doc/added "1.17"} + [cfg {:keys [id commit-id] :as params}] + ;; A save whose answer never reached the client is sent again with the + ;; same commit id; it must not be applied twice. + (let [key (some->> commit-id (commit-key id)) + claimed? (volatile! false) + result (try + (db/tx-run! cfg update-file-tx params key claimed?) + (catch Throwable cause + (when @claimed? + (release-commit! cfg key)) + (throw cause)))] + ;; Recorded once the transaction commits, so the record never vouches + ;; for changes that were rolled back. + (when @claimed? + (record-commit! cfg key (:revn result))) + result)) + +(defn- update-file-tx + "Applies a save under the file lock. A save repeating a recorded commit id + is answered without applying it again; otherwise `claimed?` is set once + the commit id is held for this save." [{:keys [::mtx/metrics ::db/conn] :as cfg} - {:keys [::rpc/profile-id id changes changes-with-metadata] :as params}] + {:keys [::rpc/profile-id id changes changes-with-metadata] :as params} + key claimed?] (files/check-edition-permissions! conn profile-id id) (db/xact-lock! conn id) - (let [file (get-file cfg id) - team (teams/get-team conn - :profile-id profile-id - :team-id (:team-id file)) + (if-let [result (some->> key (get-repeated-commit-result cfg params))] + result + (let [_ (when key + (claim-commit! cfg key) + (vreset! claimed? true)) - features (-> (cfeat/get-team-enabled-features cf/flags team) - (cfeat/check-client-features! (:features params)) - (cfeat/check-file-features! (:features file))) + file (get-file cfg id) + team (teams/get-team conn + :profile-id profile-id + :team-id (:team-id file)) - changes (if changes-with-metadata - (->> changes-with-metadata (mapcat :changes) vec) - (vec changes)) + features (-> (cfeat/get-team-enabled-features cf/flags team) + (cfeat/check-client-features! (:features params)) + (cfeat/check-file-features! (:features file))) - params (-> params - (assoc :profile-id profile-id) - (assoc :features (set/difference features cfeat/frontend-only-features)) - (assoc :team team) - (assoc :file file) - (assoc :changes changes)) + changes (if changes-with-metadata + (->> changes-with-metadata (mapcat :changes) vec) + (vec changes)) - cfg (assoc cfg ::timestamp (ct/now)) + params (-> params + (assoc :profile-id profile-id) + (assoc :features (set/difference features cfeat/frontend-only-features)) + (assoc :team team) + (assoc :file file) + (assoc :changes changes)) - tpoint (ct/tpoint)] + cfg (assoc cfg ::timestamp (ct/now)) - (when (not= (:vern params) - (:vern file)) - (ex/raise :type :validation - :code :vern-conflict - :hint "A different version has been restored for the file." - :context {:incoming-revn (:revn params) - :stored-revn (:revn file)})) + tpoint (ct/tpoint)] - (when (> (:revn params) - (:revn file)) - (ex/raise :type :validation - :code :revn-conflict - :hint "The incoming revision number is greater that stored version." - :context {:incoming-revn (:revn params) - :stored-revn (:revn file)})) + (when (not= (:vern params) + (:vern file)) + (ex/raise :type :validation + :code :vern-conflict + :hint "A different version has been restored for the file." + :context {:incoming-revn (:revn params) + :stored-revn (:revn file)})) - ;; When newly computed features does not match exactly with the - ;; features defined on team row, we update it - (when-let [features (-> features - (set/difference (:features team)) - (set/difference cfeat/no-team-inheritable-features) - (not-empty))] - (let [features (-> features - (set/union (:features team)) - (set/difference cfeat/no-team-inheritable-features) - (into-array))] - (db/update! conn :team - {:features features} - {:id (:id team)} - {::db/return-keys false}))) + (when (> (:revn params) + (:revn file)) + (ex/raise :type :validation + :code :revn-conflict + :hint "The incoming revision number is greater that stored version." + :context {:incoming-revn (:revn params) + :stored-revn (:revn file)})) + + ;; When newly computed features does not match exactly with the + ;; features defined on team row, we update it + (when-let [features (-> features + (set/difference (:features team)) + (set/difference cfeat/no-team-inheritable-features) + (not-empty))] + (let [features (-> features + (set/union (:features team)) + (set/difference cfeat/no-team-inheritable-features) + (into-array))] + (db/update! conn :team + {:features features} + {:id (:id team)} + {::db/return-keys false}))) - (mtx/run! metrics {:id :update-file-changes :inc (count changes)}) + (mtx/run! metrics {:id :update-file-changes :inc (count changes)}) - (binding [l/*context* (some-> (meta params) - (get :app.http/request) - (errors/request->context))] - (-> (update-file* cfg params) - (rph/with-defer #(let [elapsed (tpoint)] - (l/trace :hint "update-file" :time (ct/format-duration elapsed)))))))) + (binding [l/*context* (some-> (meta params) + (get :app.http/request) + (errors/request->context))] + (-> (update-file* cfg params) + (rph/with-defer #(let [elapsed (tpoint)] + (l/trace :hint "update-file" :time (ct/format-duration elapsed))))))))) (defn- update-file* "Internal function, part of the update-file process, that encapsulates @@ -459,6 +511,108 @@ (mapv (fn [row] (update row :changes blob/decode))))) +(defn- commit-key + "Redis key of a commit id. A commit id is only unique within one file, and + the tenant keeps servers sharing a redis instance apart." + [file-id commit-id] + (str "penpot.file-commit." (cf/get :tenant) "." file-id "." commit-id)) + +(defn- with-commit-store + "Runs `f` with a redis connection. Without redis the save goes ahead + unrecorded, which only loses the protection against repeats." + [cfg hint key f] + (try + (rds/run! cfg (fn [{:keys [::rds/conn]}] (f conn))) + (catch Exception cause + (l/warn :hint hint :key key :cause cause) + nil))) + +(defn- claim-commit! + "Holds the commit id for the save about to run. Raises when another save + already holds it." + [cfg key] + (let [result (with-commit-store cfg "unable to claim commit id" key + (fn [conn] + (or (rds/set conn key pending-commit + {:nx true :ex commit-pending-ttl}) + ::held)))] + (when (= ::held result) + (ex/raise :type :validation + :code :commit-in-progress + :hint "Another save with this commit id is in progress.")))) + +(defn- release-commit! + [cfg key] + (with-commit-store cfg "unable to release commit id" key + (fn [conn] (rds/del conn key)))) + +(defn- record-commit! + "Remembers the revision a save answered with, for its repeats to answer." + [cfg key revn] + (with-commit-store cfg "unable to record commit id" key + (fn [conn] (rds/set conn key (str revn) {:ex commit-retention})))) + +(defn- await-commit-revn + "The revision recorded for a commit id, or nil when no save holds it. + + This runs under the file lock, so a commit id still pending belongs to a + save that has just committed or failed and is about to say so; its outcome + is awaited for a moment. Raises when it does not come in time." + [cfg key] + (loop [waited 0] + (let [value (with-commit-store cfg "unable to read commit id" key + (fn [conn] (rds/get conn key)))] + (cond + (nil? value) + nil + + (not= pending-commit value) + (parse-long value) + + (< waited commit-await-ms) + (do (Thread/sleep ^long commit-poll-ms) + (recur (+ waited commit-poll-ms))) + + :else + (ex/raise :type :validation + :code :commit-in-progress + :hint "Another save with this commit id is in progress."))))) + +(def ^:private sql:file-for-repeated-commit + "select f.vern, + f.name, + f.features, + f.project_id, + p.team_id + from file as f + inner join project as p on (p.id = f.project_id) + where f.id = ?") + +(defn- get-repeated-commit-result + "Answers a save this server already applied, under the same commit id, + without applying its changes again. + + Raises a version conflict when the file has been rolled back since: the + record outlives the change it made, and answering would make the client + discard an edit the file does not carry." + [{:keys [::db/conn] :as cfg} {:keys [id vern]} key] + (when-let [revn (await-commit-revn cfg key)] + (let [row (db/exec-one! conn [sql:file-for-repeated-commit id])] + (when (not= vern (:vern row)) + (ex/raise :type :validation + :code :vern-conflict + :hint "A different version has been restored for the file." + :context {:incoming-vern vern + :stored-vern (:vern row)})) + + (with-meta {:revn revn} + {::audit/replace-props + {:id id + :name (:name row) + :features (db/decode-pgarray (:features row) #{}) + :project-id (:project-id row) + :team-id (:team-id row)}})))) + (defn- send-notifications! [cfg {:keys [team changes session-id] :as params} file] (let [lchanges (filter library-change? changes) diff --git a/backend/test/backend_tests/rpc_file_test.clj b/backend/test/backend_tests/rpc_file_test.clj index 3ebcc4b80f..cc6847cf44 100644 --- a/backend/test/backend_tests/rpc_file_test.clj +++ b/backend/test/backend_tests/rpc_file_test.clj @@ -7,6 +7,7 @@ (ns backend-tests.rpc-file-test (:require [app.common.features :as cfeat] + [app.common.generic-pool :as gpool] [app.common.pprint :as pp] [app.common.thumbnails :as thc] [app.common.time :as ct] @@ -17,12 +18,15 @@ [app.db.sql :as sql] [app.features.fdata :as fdata] [app.http :as http] + [app.redis :as rds] [app.rpc :as-alias rpc] [app.rpc.commands.files :as files] [app.storage :as sto] [backend-tests.helpers :as th] [clojure.test :as t] - [cuerdas.core :as str])) + [cuerdas.core :as str]) + (:import + java.lang.AutoCloseable)) (t/use-fixtures :once th/state-init) (t/use-fixtures :each th/database-reset) @@ -2672,3 +2676,278 @@ (t/is (th/ex-info? (:error out))) (t/is (th/ex-of-type? (:error out) :validation)) (t/is (th/ex-of-code? (:error out) :params-validation)))) + +(t/deftest update-file-applies-a-repeated-commit-only-once + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + commit-id (uuid/random) + + params {::th/type :update-file + ::rpc/profile-id (:id profile) + :id (:id file) + :session-id (uuid/random) + :revn 0 + :vern 0 + :commit-id commit-id + :features cfeat/supported-features + :changes [{:type :add-page + :name "page to add once" + :id (uuid/random)}]} + + ;; The client never learns the outcome of the first call and sends + ;; the same commit again. + out1 (th/command! params) + out2 (th/command! params)] + + (t/is (nil? (:error out1))) + (t/is (nil? (:error out2))) + + (t/testing "the page is added once" + (let [out (th/command! {::th/type :get-file + ::rpc/profile-id (:id profile) + :id (:id file)})] + (t/is (nil? (:error out))) + (t/is (= 2 (count (get-in out [:result :data :pages])))))) + + (t/testing "the file advances a single revision" + (let [row (th/db-get :file {:id (:id file)})] + (t/is (= 1 (:revn row))))) + + (t/testing "the repeated call answers exactly what the first one did" + (t/is (= (:revn (:result out1)) + (:revn (:result out2))))) + + (t/testing "the repeated call skips the lagged changes nobody reads" + (t/is (not (contains? (:result out2) :lagged)))) + + (t/testing "a replay is audited like the call it repeats" + (let [props1 (-> out1 :result meta :app.loggers.audit/replace-props) + props2 (-> out2 :result meta :app.loggers.audit/replace-props)] + (t/is (= (set (keys props1)) (set (keys props2)))) + (t/is (= (:id props1) (:id props2))) + (t/is (= (:team-id props1) (:team-id props2))))))) + +(defn- save-params + "Params for one update-file call, adding a page. A nil commit-id is what a + client that knows nothing about commit ids sends. + + Call this once and reuse the result to repeat a save: a retry carries the + same changes, page id included, as the call it repeats." + [profile file commit-id page-name] + (cond-> {::th/type :update-file + ::rpc/profile-id (:id profile) + :id (:id file) + :session-id (uuid/random) + :revn 0 + :vern 0 + :features cfeat/supported-features + :changes [{:type :add-page + :name page-name + :id (uuid/random)}]} + (some? commit-id) + (assoc :commit-id commit-id))) + +(defn- count-pages + [profile file] + (let [out (th/command! {::th/type :get-file + ::rpc/profile-id (:id profile) + :id (:id file)})] + (t/is (nil? (:error out))) + (count (get-in out [:result :data :pages])))) + +(defn- file-revn + "How many times changes were applied to a file. Repeating a save carries + the same page id, so the pages look the same whether the changes were + applied once or twice; the revision is what tells them apart." + [file] + (:revn (th/db-get :file {:id (:id file)}))) + +(defn- commit-key + [file-id commit-id] + (str "penpot.file-commit." (cf/get :tenant) "." file-id "." commit-id)) + +(defn- with-redis + [f] + (let [conn (gpool/get (get-in th/*system* [:app.redis/pool]))] + (try + (f @conn) + (finally + (.close ^AutoCloseable conn))))) + +(defn- get-commit-record + [file-id commit-id] + (with-redis #(rds/get % (commit-key file-id commit-id)))) + +(t/deftest a-saved-commit-is-recorded-with-its-revision + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + commit-id (uuid/random) + out (th/command! (save-params profile file commit-id "page"))] + + (t/is (nil? (:error out))) + (t/is (= "0" (get-commit-record (:id file) commit-id)) + "the record holds the revision the save answered with"))) + +(t/deftest a-failed-save-releases-its-commit-id + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + commit-id (uuid/random) + params (save-params profile file commit-id "page") + ;; A revision ahead of the file makes the save fail after it has + ;; claimed its commit id. + out1 (th/command! (assoc params :revn 10))] + + (t/is (th/ex-of-code? (:error out1) :revn-conflict)) + (t/is (nil? (get-commit-record (:id file) commit-id)) + "nothing vouches for changes that were not applied") + + (t/testing "the corrected save applies" + (let [out2 (th/command! params)] + (t/is (nil? (:error out2))) + (t/is (= 1 (file-revn file))))))) + +(t/deftest a-repeat-of-a-save-in-progress-is-turned-away + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + commit-id (uuid/random)] + + ;; A save that holds the commit id and never records its outcome. + (with-redis #(rds/set % (commit-key (:id file) commit-id) "pending" + {:ex (ct/duration {:minutes 1})})) + + (let [out (th/command! (save-params profile file commit-id "page"))] + (t/is (th/ex-of-type? (:error out) :validation)) + (t/is (th/ex-of-code? (:error out) :commit-in-progress)) + (t/is (= 0 (file-revn file)) "nothing is applied")))) + +(t/deftest update-file-refuses-a-repeated-commit-after-a-version-restore + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + commit-id (uuid/random) + + params {::th/type :update-file + ::rpc/profile-id (:id profile) + :id (:id file) + :session-id (uuid/random) + :revn 0 + :vern 0 + :commit-id commit-id + :features cfeat/supported-features + :changes [{:type :add-page + :name "page to add once" + :id (uuid/random)}]} + + out1 (th/command! params)] + + (t/is (nil? (:error out1))) + + ;; Restoring a version assigns the file a fresh vern and rolls its data + ;; back, so the effect of the commit is gone even though its id remains. + (th/db-update! :file {:vern 7} {:id (:id file)}) + + (let [out2 (th/command! params) + edata (-> out2 :error ex-data)] + (t/testing "a version conflict is raised" + (t/is (some? (:error out2))) + (t/is (= :validation (:type edata))) + (t/is (= :vern-conflict (:code edata))))))) + +(t/deftest a-save-without-a-commit-id-is-never-deduplicated + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + params (save-params profile file nil "page") + + out1 (th/command! params) + out2 (th/command! params)] + + (t/is (nil? (:error out1))) + (t/is (nil? (:error out2))) + + ;; The dedup must not reach a client that sends no commit id. + (t/testing "both calls apply" + (t/is (= 2 (file-revn file)))))) + +(t/deftest two-commits-carrying-the-same-changes-both-apply + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + params (save-params profile file (uuid/random) "page") + + out1 (th/command! params) + out2 (th/command! (assoc params :commit-id (uuid/random)))] + + (t/is (nil? (:error out1))) + (t/is (nil? (:error out2))) + + ;; The dedup keys on the commit id, not on what the changes look like. + (t/testing "both calls apply" + (t/is (= 2 (file-revn file)))))) + +(t/deftest one-commit-id-used-against-two-files-applies-to-both + (let [profile (th/create-profile* 1 {:is-active true}) + file1 (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + file2 (th/create-file* 2 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + commit-id (uuid/random) + + out1 (th/command! (save-params profile file1 commit-id "page")) + out2 (th/command! (save-params profile file2 commit-id "page"))] + + (t/is (nil? (:error out1))) + (t/is (nil? (:error out2))) + + ;; A commit id is only ever unique within one file. + (t/testing "neither file is deduplicated against the other" + (t/is (= 2 (count-pages profile file1))) + (t/is (= 2 (count-pages profile file2)))))) + +(t/deftest concurrent-repeats-of-one-commit-apply-once + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile) + :is-shared false}) + + params (save-params profile file (uuid/random) "page") + + ;; A client that retries before the first answer arrives has two + ;; identical requests in flight at once. The file advisory lock + ;; serialises them, and the second one waits for the outcome the + ;; first records. The latch holds both threads until they can start + ;; together. + start (java.util.concurrent.CountDownLatch. 1) + pending (doall [(future (.await start) (th/command! params)) + (future (.await start) (th/command! params))]) + _ (.countDown start) + outs (mapv #(deref % 30000 ::timeout) pending)] + + (t/is (not-any? #{::timeout} outs) "both requests finished") + (t/is (every? (comp nil? :error) outs)) + + (t/testing "the changes are applied once" + (t/is (= 2 (count-pages profile file))) + (t/is (= 1 (file-revn file)))) + + (t/testing "both callers get the same answer" + (t/is (= 0 (:revn (:result (first outs))))) + (t/is (= 0 (:revn (:result (second outs))))))))