diff --git a/deps.edn b/deps.edn index 47ad5ed..ec80312 100644 --- a/deps.edn +++ b/deps.edn @@ -1,6 +1,6 @@ {:paths ["src" "test"] :deps {org.clojure/clojure {:mvn/version "1.12.0"} - io.github.radarroark/xitdb {:mvn/version "0.37.0"}} + io.github.radarroark/xitdb {:mvn/version "0.38.0"}} :aliases {:test {:extra-deps {io.github.cognitect-labs/test-runner diff --git a/src/xitdb/db.clj b/src/xitdb/db.clj index 4aefd78..43341b2 100644 --- a/src/xitdb/db.clj +++ b/src/xitdb/db.clj @@ -88,6 +88,14 @@ cursor (conversion/keypath-cursor root-cursor base-keypath)] (.write cursor (conversion/v->slot! cursor retval))))))) +(defn- with-file-lock [^Database db f] + (let [core (.-core db)] + (if (instance? CoreBufferedFile core) + (with-open [held (or (.tryLock (.getChannel (.-file ^CoreBufferedFile core))) + (throw (IllegalStateException. "Database is locked for writing")))] + (f)) + (f)))) + (defn xitdb-swap-with-lock! "Performs the 'swap!' operation while locking `db.lock`. Returns the new value of the database. @@ -102,12 +110,14 @@ (throw (IllegalStateException. "swap! should not be called from swap! or reset!"))) (try (.lock lock) - (let [old-value (when *return-history?* (deref xitdb)) - index (apply xitdb-swap! (into [(-> xitdb .rwdb) base-keypath f] args)) - new-value (deref xitdb)] - (if *return-history?* - [index old-value new-value] - new-value)) + (with-file-lock (.-rwdb xitdb) + (fn [] + (let [old-value (when *return-history?* (deref xitdb)) + index (apply xitdb-swap! (into [(-> xitdb .rwdb) base-keypath f] args)) + new-value (deref xitdb)] + (if *return-history?* + [index old-value new-value] + new-value)))) (finally (.unlock lock))))) @@ -164,9 +174,11 @@ (try (.lock lock) - (let [history (db-history rwdb)] - (xitdb-reset! history new-value) - (deref this)) + (with-file-lock rwdb + (fn [] + (let [history (db-history rwdb)] + (xitdb-reset! history new-value) + (deref this)))) (finally (.unlock lock)))) diff --git a/test/xitdb/file_lock_test.clj b/test/xitdb/file_lock_test.clj new file mode 100644 index 0000000..4ac49c6 --- /dev/null +++ b/test/xitdb/file_lock_test.clj @@ -0,0 +1,62 @@ +(ns xitdb.file-lock-test + (:require [clojure.java.io :as io] + [clojure.test :refer :all] + [xitdb.db :as db]) + (:import [java.util.concurrent TimeUnit])) + +(defn- with-temp-db [f] + (let [file (java.io.File/createTempFile "xitdb-lock-" ".db")] + (try (f (.getAbsolutePath file)) (finally (.delete file))))) + +(deftest overlapping-writes-are-rejected-and-locks-are-released + (with-temp-db + (fn [path] + (with-open [a (db/xit-db path) + b (db/xit-db path)] + (reset! a {:initial 1}) + (swap! a (fn [m] + (is (thrown? IllegalStateException + (swap! b (fn [_] (is false "callback must not run"))))) + (is (thrown? IllegalStateException (reset! b {:lost true}))) + (is (thrown? IllegalStateException + (reset! (db/xdb-cursor b [:initial]) 99))) + (assoc m :a 3))) + (swap! b assoc :b 2) + (is (thrown? clojure.lang.ExceptionInfo + (swap! a (fn [m] + (assoc m :aborted [1 2 3]) + (throw (ex-info "abort" {})))))) + (reset! b (db/materialize @b))) + (with-open [reopened (db/xit-db path)] + (is (= {:initial 1 :a 3 :b 2} (db/materialize @reopened))) + (is (= 4 (count reopened))))))) + +(deftest foreign-process-lock-prevents-swap-and-reset + (with-temp-db + (fn [path] + (with-open [d (db/xit-db path)] + (reset! d {:safe 1}) + (let [code (str "(with-open [f (java.io.RandomAccessFile. " (pr-str path) + " \"rw\") held (.tryLock (.getChannel f))]" + " (println (boolean held)) (flush) (read-line))") + process (.start (ProcessBuilder. + ^java.util.List + [(str (System/getProperty "java.home") "/bin/java") + "-cp" (System/getProperty "java.class.path") + "clojure.main" "-e" code]))] + (try + (with-open [reader (io/reader (.getInputStream process)) + writer (io/writer (.getOutputStream process))] + (is (= "true" (deref (future (.readLine reader)) 10000 ::timeout))) + (is (thrown? IllegalStateException (swap! d assoc :lost true))) + (is (thrown? IllegalStateException (reset! d {}))) + (is (= {:safe 1} (db/materialize @d))) + (is (= 1 (count d))) + (.write writer "\n") + (.flush writer)) + (finally + (when-not (.waitFor process 5 TimeUnit/SECONDS) + (.destroyForcibly process) + (.waitFor process 5 TimeUnit/SECONDS))))) + (swap! d assoc :after true) + (is (= {:safe 1 :after true} (db/materialize @d)))))))