Repository: CSCfi/rems
Pull request #1153 (2019-04-23 14:01:18)

Transaction stuff
- Make also reads transactional
- Test that events are observed in monotonically increasing order


Patch:
diff --git a/resources/sql/queries.sql b/resources/sql/queries.sql
index 55a5b873f..9267822cf 100644
--- a/resources/sql/queries.sql
+++ b/resources/sql/queries.sql
@@ -483,7 +483,7 @@ SET
   --~(when (contains? params :archived) ", archived = :archived")
   --~(when (contains? params :title) ", title = :title")
   --~(when (contains? params :workflow) ", workflowBody = :workflow::jsonb")
-WHERE id = :id
+WHERE id = :id;
 
 -- :name create-workflow-license! :insert
 INSERT INTO workflow_licenses
@@ -727,7 +727,8 @@ SELECT prefix, suffix FROM external_application_id
 /*~ (when (:prefix params) */
 WHERE prefix = :prefix
 /*~ ) ~*/
+;
 
 -- :name add-external-id! :!
 INSERT INTO external_application_id (prefix, suffix)
-VALUES (:prefix, :suffix)
+VALUES (:prefix, :suffix);
diff --git a/src/clj/rems/api.clj b/src/clj/rems/api.clj
index aba049d1f..f90db072d 100644
--- a/src/clj/rems/api.clj
+++ b/src/clj/rems/api.clj
@@ -64,14 +64,13 @@
     :access-control-allow-origin #".*"
     :access-control-allow-methods [:get :put :post :delete]))
 
-(defn- should-wrap-transaction? [request]
-  (contains? #{:put :post} (:request-method request)))
+(defn- read-only? [request]
+  (not (contains? #{:put :post} (:request-method request))))
 
 (defn transaction-middleware [handler]
   (fn [request]
-    (if (should-wrap-transaction? request)
-      (conman/with-transaction [rems.db.core/*db* {:isolation :serializable}]
-        (handler request))
+    (conman/with-transaction [rems.db.core/*db* {:isolation :serializable
+                                                 :read-only? (read-only? request)}]
       (handler request))))
 
 (defn slow-middleware [request]
diff --git a/src/clj/rems/application/events_cache.clj b/src/clj/rems/application/events_cache.clj
index f0b9952d3..2e5b26b5e 100644
--- a/src/clj/rems/application/events_cache.clj
+++ b/src/clj/rems/application/events_cache.clj
@@ -1,5 +1,7 @@
 (ns rems.application.events-cache
-  (:require [rems.db.applications :as applications]
+  (:require [conman.core :as conman]
+            [rems.db.applications :as applications]
+            [rems.db.core :refer [*db*]]
             [rems.util :refer [atom?]]))
 
 (def ^:private empty-cache
@@ -15,17 +17,19 @@
    `update-fn` should be a function which takes as parameters the previously
    cached state and a list of new events, and returns the updated state."
   [cache update-fn]
-  (let [cache-enabled? (atom? cache)
-        cached (if cache-enabled? @cache empty-cache)
-        new-events (applications/get-all-events-since (:last-processed-event-id cached))]
-    (if (empty? new-events)
-      (:state cached)
-      (let [updated {:state (update-fn (:state cached) new-events)
-                     :last-processed-event-id (:event/id (last new-events))}]
-        ;; with concurrent requests, only one of them will update the cache
-        (when cache-enabled?
-          (compare-and-set! cache cached updated))
-        (:state updated)))))
+  (conman/with-transaction [*db* {:isolation :serializable
+                                  :read-only? true}]
+    (let [cache-enabled? (atom? cache)
+          cached (if cache-enabled? @cache empty-cache)
+          new-events (applications/get-all-events-since (:last-processed-event-id cached))]
+      (if (empty? new-events)
+        (:state cached)
+        (let [updated {:state (update-fn (:state cached) new-events)
+                       :last-processed-event-id (:event/id (last new-events))}]
+          ;; with concurrent requests, only one of them will update the cache
+          (when cache-enabled?
+            (compare-and-set! cache cached updated))
+          (:state updated))))))
 
 (defn empty! [cache]
   (when (atom? cache)
diff --git a/src/clj/rems/db/applications.clj b/src/clj/rems/db/applications.clj
index c326ae918..279e058a7 100644
--- a/src/clj/rems/db/applications.clj
+++ b/src/clj/rems/db/applications.clj
@@ -2,12 +2,13 @@
   "Query functions for forms and applications."
   (:require [clj-time.core :as time]
             [clj-time.format :as time-format]
+            [clojure.java.jdbc :as jdbc]
             [clojure.test :refer [deftest is]]
             [conman.core :as conman]
             [cprop.tools :refer [merge-maps]]
+            [rems.application-util :refer [form-fields-editable?]]
             [rems.application.events :as events]
             [rems.application.model :as model]
-            [rems.application-util :refer [form-fields-editable?]]
             [rems.auth.util :refer [throw-forbidden]]
             [rems.db.catalogue :as catalogue]
             [rems.db.core :as db]
@@ -750,6 +751,11 @@
 
 (defn command! [cmd]
   (assert (:application-id cmd))
+  ;; Use locks to prevent multiple commands being executed in parallel.
+  ;; Serializable isolation level will already avoid anomalies, but produces
+  ;; lots of transaction conflicts when there is contention. This lock
+  ;; roughly doubles the throughput for rems.db.test-transactions tests.
+  (jdbc/execute! db/*db* ["LOCK TABLE application_event IN SHARE ROW EXCLUSIVE MODE"])
   (let [events (get-application-events (:application-id cmd))
         app (-> nil
                 (dynamic/apply-events events)
diff --git a/src/clj/rems/workflow/dynamic.clj b/src/clj/rems/workflow/dynamic.clj
index 6fb66d1dd..522b24209 100644
--- a/src/clj/rems/workflow/dynamic.clj
+++ b/src/clj/rems/workflow/dynamic.clj
@@ -249,12 +249,16 @@
     (update result :result add-common-event-fields-from-command cmd)
     result))
 
+(defn ^:dynamic postprocess-command-result-for-tests [result _cmd _application]
+  result)
+
 (defn handle-command [cmd application injections]
   (commands/validate-command cmd) ; this is here mostly for tests, commands via the api are validated by compojure-api
   (let [permissions (permissions/user-permissions application (:actor cmd))]
     (if (contains? permissions (:type cmd))
       (-> (command-handler cmd application injections)
-          (enrich-result cmd))
+          (enrich-result cmd)
+          (postprocess-command-result-for-tests cmd application))
       {:errors (or (:errors (command-handler cmd application injections)) ; prefer more specific error
                    [{:type :forbidden}])})))
 
diff --git a/src/cljs/rems/administration/create_form.cljs b/src/cljs/rems/administration/create_form.cljs
index 5757a6a76..06e50fc92 100644
--- a/src/cljs/rems/administration/create_form.cljs
+++ b/src/cljs/rems/administration/create_form.cljs
@@ -93,10 +93,8 @@
 
 (defn- valid-request-field? [field languages]
   (and (valid-required-localized-string? (:title field) languages)
+       (boolean? (:optional field))
        (not (str/blank? (:type field)))
-       (if (supports-optional? field)
-         (boolean? (:optional field))
-         (nil? (:optional field)))
        (if (supports-input-prompt? field)
          (valid-optional-localized-string? (:input-prompt field) languages)
          (nil? (:input-prompt field)))
@@ -118,9 +116,8 @@
 
 (defn- build-request-field [field languages]
   (merge {:title (build-localized-string (:title field) languages)
+          :optional (boolean (:optional field))
           :type (:type field)}
-         (when (supports-optional? field)
-           {:optional (boolean (:optional field))})
          (when (supports-input-prompt? field)
            {:input-prompt (build-localized-string (:input-prompt field) languages)})
          (when (supports-maxlength? field)
@@ -285,16 +282,12 @@
 (defn- form-field-to-application-field
   "Convert a field from the form create model to the application view model."
   [field]
-  (merge {:field/type (keyword (:type field))
-          :field/title (:title field)}
-         (when (supports-optional? field)
-           {:field/optional (:optional field)})
-         (when (supports-input-prompt? field)
-           {:field/placeholder (:input-prompt field)})
-         (when (supports-maxlength? field)
-           {:field/max-length (parse-maxlength (:maxlength field))})
-         (when (supports-options? field)
-           {:field/options (:options field)})))
+  {:field/type (keyword (:type field))
+   :field/title (:title field)
+   :field/placeholder (:input-prompt field)
+   :field/max-length (parse-maxlength (:maxlength field))
+   :field/optional (:optional field)
+   :field/options (:options field)})
 
 (defn- field-preview [field]
   [fields/field (form-field-to-application-field field)])


Test patch:
diff --git a/test/clj/rems/db/test_transactions.clj b/test/clj/rems/db/test_transactions.clj
new file mode 100644
index 000000000..cdfd4bc11
--- /dev/null
+++ b/test/clj/rems/db/test_transactions.clj
@@ -0,0 +1,189 @@
+(ns ^:integration rems.db.test-transactions
+  (:require [clj-time.core :as time]
+            [clojure.test :refer :all]
+            [conman.core :as conman]
+            [rems.db.applications :as applications]
+            [rems.db.catalogue :as catalogue]
+            [rems.db.core :as db]
+            [rems.db.form :as form]
+            [rems.db.resource :as resource]
+            [rems.db.testing :refer [test-db-fixture rollback-db-fixture test-data-fixture]]
+            [rems.db.users :as users]
+            [rems.db.workflow :as workflow]
+            [rems.workflow.dynamic :as dynamic])
+  (:import [java.sql SQLException]
+           [java.util UUID]
+           [java.util.concurrent Executors Future TimeUnit ExecutorService]
+           [org.postgresql.util PSQLException]))
+
+(use-fixtures
+  :once
+  test-db-fixture)
+
+(defn- create-dummy-user []
+  (let [user-id "user"]
+    (users/add-user! user-id {:eppn user-id})
+    user-id))
+
+(defn- create-dummy-application [user-id]
+  (let [workflow-id (:id (workflow/create-workflow! {:user-id user-id
+                                                     :organization ""
+                                                     :title ""
+                                                     :type :dynamic
+                                                     :handlers []}))
+        form-id (:id (form/create-form! user-id
+                                        {:organization ""
+                                         :title ""
+                                         :fields []}))
+        res-id (:id (resource/create-resource! {:resid (str "urn:uuid:" (UUID/randomUUID))
+                                                :organization ""
+                                                :licenses []}
+                                               user-id))
+        cat-id (:id (catalogue/create-catalogue-item! {:title ""
+                                                       :form form-id
+                                                       :resid res-id
+                                                       :wfid workflow-id}))
+        app-id (:application-id (applications/create-application! user-id [cat-id]))]
+    (assert app-id)
+    app-id))
+
+(defn- transaction-conflict? [^Exception e]
+  (cond
+    (nil? e) false
+    (instance? PSQLException e) (.contains (.getMessage e)
+                                           "The transaction might succeed if retried")
+    :else (transaction-conflict? (.getCause e))))
+
+(defn- sample-until-interrupted [f]
+  (loop [results []]
+    (if (.isInterrupted (Thread/currentThread))
+      results
+      (recur (try
+               (conj results (f))
+               (catch InterruptedException _
+                 (.interrupt (Thread/currentThread))
+                 results)
+               ;; XXX: HikariPool.getConnection wraps InterruptedException into SQLException
+               (catch SQLException e
+                 (if (instance? InterruptedException (.getCause e))
+                   (do (.interrupt (Thread/currentThread))
+                       results)
+                   (throw e))))))))
+
+(defn- submit-all [^ExecutorService thread-pool tasks]
+  (doall (for [task tasks]
+           (.submit thread-pool ^Callable task))))
+
+(deftest test-event-publishing-consistency
+  (let [test-duration-millis 2000
+        applications-count 5
+        writers-per-application 5
+        concurrent-readers 5
+        user-id (create-dummy-user)
+        app-ids (vec (for [_ (range applications-count)]
+                       (create-dummy-application user-id)))
+        ;; Currently we only test that commands are not executed concurrently
+        ;; for a single application. To guarantee that, we could add an app version
+        ;; column to the events table with constraint `UNIQUE (appId, appVersion)`.
+        observed-app-version-marker 999
+        mark-observed-app-version (fn [result _cmd application]
+                                    (if (and (:success result)
+                                             (= :application.event/draft-saved (:event/type (:result result))))
+                                      (assoc-in result [:result :application/field-values observed-app-version-marker]
+                                                (str (count (:application/events application))))
+                                      result))
+        write-event (fn [app-id]
+                      (try
+                        (conman/with-transaction [db/*db* {:isolation :serializable}]
+                          (binding [dynamic/postprocess-command-result-for-tests mark-observed-app-version]
+                            (applications/command!
+                             {:type :application.command/save-draft
+                              :time (time/now)
+                              :actor user-id
+                              :application-id app-id
+                              :field-values []})))
+                        (catch Exception e
+                          (if (transaction-conflict? e)
+                            ::transaction-conflict
+                            (throw e)))))
+        read-app-events (fn [app-id]
+                          (conman/with-transaction [db/*db* {:isolation :serializable
+                                                             :read-only? true}]
+                            {::app-id app-id
+                             ::events (applications/get-application-events app-id)}))
+        read-all-events (fn []
+                          (conman/with-transaction [db/*db* {:isolation :serializable
+                                                             :read-only? true}]
+                            {::events (applications/get-all-events-since 0)}))
+        thread-pool (Executors/newCachedThreadPool)
+        app-events-readers (submit-all thread-pool (for [app-id app-ids]
+                                                     (fn [] (sample-until-interrupted
+                                                             (fn [] (read-app-events app-id))))))
+        all-events-readers (submit-all thread-pool (for [_ (range concurrent-readers)]
+                                                     (fn [] (sample-until-interrupted
+                                                             (fn [] (read-all-events))))))
+        writers (submit-all thread-pool (for [app-id app-ids
+                                              _ (range writers-per-application)]
+                                          (fn [] (sample-until-interrupted
+                                                  (fn [] (write-event app-id))))))]
+    (Thread/sleep test-duration-millis)
+    (doto thread-pool
+      (.shutdownNow)
+      (.awaitTermination 30 TimeUnit/SECONDS))
+
+    (let [writer-attempts (flatten (map #(.get ^Future %) writers))
+          writer-results (remove #{::transaction-conflict} writer-attempts)
+          app-events-reader-results (flatten (map #(.get ^Future %) app-events-readers))
+          all-events-reader-results (flatten (map #(.get ^Future %) all-events-readers))
+          final-events (applications/get-all-events-since 0)
+          final-events-by-app-id (group-by :application/id final-events)]
+      (comment
+        (prn 'writer-attempts (count writer-attempts))
+        (prn 'writer-results (count writer-results))
+        (prn 'app-events-reader-results (count app-events-reader-results))
+        (prn 'all-events-reader-results (count all-events-reader-results)))
+
+      (testing "all commands succeeded"
+        (is (not (empty? writer-results)))
+        (is (every? nil? writer-results)) ; successful commands return nil
+        (is (= (count writer-results)
+               (count writer-attempts))
+            "should have no transaction conflicts"))
+
+      (testing "all events were written"
+        (is (= (+ (count app-ids) ; one application created event per application
+                  (count writer-results))
+               (count final-events))))
+
+      (testing "event IDs are observed in monotonically increasing order, within one application"
+        (doseq [observed app-events-reader-results]
+          (let [final-events (get final-events-by-app-id (::app-id observed))]
+            (is (= (->> final-events
+                        (take (count (::events observed)))
+                        (map :event/id))
+                   (->> (::events observed)
+                        (map :event/id)))))))
+
+      (testing "event IDs are observed in monotonically increasing order, globally"
+        (doseq [observed all-events-reader-results]
+          (is (= (->> final-events
+                      (take (count (::events observed)))
+                      (map :event/id))
+                 (->> (::events observed)
+                      (map :event/id))))))
+
+      (testing "commands are executed serially"
+        (doseq [[_ events] final-events-by-app-id]
+          (is (= (->> (range 1 (count events))
+                      (map str))
+                 (->> events
+                      (filter #(= :application.event/draft-saved (:event/type %)))
+                      (map #(get-in % [:application/field-values observed-app-version-marker])))))))
+
+      (testing "there are not gaps in event IDs"
+        ;; There might still be gaps in the IDs if the transaction is rolled back
+        ;; for an unrelated reason after the command is executed. To guarantee no gaps,
+        ;; the event IDs could be generated with `max(id) + 1`.
+        (is (every? (fn [[a b]] (= (inc a) b))
+                    (->> (map :event/id final-events)
+                         (partition 2 1))))))))