✨ Apply a repeated update-file only once

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)

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