diff --git a/.clj-kondo/config.edn b/.clj-kondo/config.edn index 139d42b..5d8a9ef 100644 --- a/.clj-kondo/config.edn +++ b/.clj-kondo/config.edn @@ -1,5 +1,12 @@ {:lint-as {duct.test/with-system clojure.core/with-open + linear.adapter.slatedb.connection/with-database next.jdbc/with-transaction + linear.adapter.slatedb.ffi/with-snapshot next.jdbc/with-transaction + linear.adapter.slatedb.ffi/with-transaction next.jdbc/with-transaction + linear.adapter.slatedb.ffi/with-change-notifications next.jdbc/with-transaction + linear.adapter.sqlite.connection/with-statement next.jdbc/with-transaction + linear.adapter.sqlite.evaluator/with-evaluation next.jdbc/with-transaction + linear.adapter.sqlite.connection-test/with-connection clojure.core/fn linear.usecase.database/with-database next.jdbc/with-transaction linear.usecase.database.revisions/with-consistent-view next.jdbc/with-transaction linear.usecase.vault/with-transaction next.jdbc/with-transaction diff --git a/compose.yaml b/compose.yaml index d512cde..47e3fd7 100644 --- a/compose.yaml +++ b/compose.yaml @@ -27,6 +27,7 @@ services: JDBC_DATABASE_URL: jdbc:postgresql://postgres:5432/linear?user=postgres&password=postgres SLATEDB_OBJECT_STORE_URL: memory:/// REVISION_RETENTION_DAYS: + MAX_LONG_POLL_TIMEOUT_MS: LINEAR_DEPLOYMENT_TARGET: GCP_KMS_KEY_NAME: OIDC_ISSUER: diff --git a/deps.edn b/deps.edn index 684bb37..2f0240e 100644 --- a/deps.edn +++ b/deps.edn @@ -17,7 +17,7 @@ com.google.cloud/google-cloud-kms {:mvn/version "2.102.0"} camel-snake-kebab/camel-snake-kebab {:mvn/version "0.4.3"} diehard/diehard {:mvn/version "0.12.1"} - io.slatedb/slatedb-uniffi {:mvn/version "0.17.0"} + io.slatedb/slatedb-uniffi {:mvn/version "0.18.0"} org.roaringbitmap/RoaringBitmap {:mvn/version "1.6.25"} funcool/promesa {:mvn/version "12.0.1"} com.github.hden/labrador {:mvn/version "0.2.0"} diff --git a/docs/turso-compatibility.md b/docs/turso-compatibility.md index 779cb15..8836d5a 100644 --- a/docs/turso-compatibility.md +++ b/docs/turso-compatibility.md @@ -12,7 +12,8 @@ E2E scenarios, run by [CI](../.github/workflows/ci.yml) on Linux x64. | Retry after a lost push response | Supported; applied progress prevents duplicate SQL execution | | Atomic application data and sync progress | Supported; committed in the same revision | | SQL failure | Entire batch rolls back; COMMIT reports `TRANSACTION_ROLLED_BACK` | -| Query partial sync and long polling | Unsupported | +| Long polling | Supported | +| Query partial sync | Unsupported | | MVCC and experimental remote writes | Outside the verified compatibility scope | ## Limitations @@ -42,7 +43,6 @@ implementation or verification work, not passing claims for Linear: | [Python `test_partial_sync`](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/python/tests/test_database_sync.py) | Query bootstrap; page selection via `server_pages_selector` is implemented locally. | | [JavaScript partial-sync tests](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/javascript/sync/packages/native/promise.test.ts) | Partial-sync concurrency, segment size, and prefetch. | | [Python `test_pull_bytes_threshold`](https://github.com/tursodatabase/turso/blob/v0.8.3/bindings/python/tests/test_database_sync.py) | Upstream edge cases beyond the repeated selector-based pulls covered by E2E. | -| [`PullUpdatesReqProtoBody.long_poll_timeout_ms`](https://github.com/tursodatabase/turso/blob/v0.8.3/sync/engine/src/server_proto.rs) | Long-poll wake-up on a newly published revision. | Upstream includes a [CLI sync server](https://github.com/tursodatabase/turso/blob/v0.8.3/cli/sync_server.rs) with page-selector support. Its JavaScript partial-sync tests skip that server diff --git a/duct.edn b/duct.edn index 7aca0da..5da9e25 100644 --- a/duct.edn +++ b/duct.edn @@ -11,6 +11,9 @@ revision-retention-days {:env "REVISION_RETENTION_DAYS" :type :int :default 30} + max-long-poll-timeout-ms {:env "MAX_LONG_POLL_TIMEOUT_MS" + :type :int + :default 5000} sqlite-vfs-name {:env "SQLITE_VFS_NAME" :default "linear-sqlite"} deployment-target {:env "LINEAR_DEPLOYMENT_TARGET" @@ -46,9 +49,11 @@ :url #ig/var oidc-jwks-url} :linear.server.jetty/allow-encoded-path-separators {} + :linear.server.jetty/thread-pool {} :duct.server.http/jetty - {:configurator #ig/ref :linear.server.jetty/allow-encoded-path-separators} + {:configurator #ig/ref :linear.server.jetty/allow-encoded-path-separators + :thread-pool #ig/ref :linear.server.jetty/thread-pool} :duct.module/web {:features #{:api} :handler-opts @@ -56,7 +61,8 @@ :linear.usecase.core/database #ig/ref :linear.adapter.postgres/datasource :linear.usecase.core/revision-store #ig/ref :linear.adapter.slatedb.store/store :linear.usecase.core/evaluator #ig/ref :linear.adapter.sqlite.evaluator/evaluator - :linear.usecase.core/key-service #ig/ref :linear.adapter.crypto/key-service} + :linear.usecase.core/key-service #ig/ref :linear.adapter.crypto/key-service + :linear.usecase.core/max-long-poll-timeout-ms #ig/var max-long-poll-timeout-ms} :middleware-opts {:provider #ig/ref :duct.auth.jwks/provider :issuer #ig/var oidc-issuer diff --git a/src/linear/adapter/slatedb/connection.clj b/src/linear/adapter/slatedb/connection.clj index 41a6f71..7b3e462 100644 --- a/src/linear/adapter/slatedb/connection.clj +++ b/src/linear/adapter/slatedb/connection.clj @@ -1,60 +1,43 @@ (ns linear.adapter.slatedb.connection (:require - [linear.adapter.slatedb.ffi :as ffi]) + [linear.adapter.slatedb.ffi :as ffi] + [promesa.core :as p] + [promesa.exec :as px]) (:import - (java.lang AutoCloseable) (org.apache.commons.pool2 BaseKeyedPooledObjectFactory PooledObject) (org.apache.commons.pool2.impl DefaultPooledObject GenericKeyedObjectPool GenericKeyedObjectPoolConfig))) -(defrecord Connection - [borrow-database return-database close-resources revision-retention-days]) +(defrecord Connection [pool object-store revision-retention-days]) -(defn connection [options] - (map->Connection options)) - -(defrecord ^:private DatabaseLease [raw-database release!] - AutoCloseable - (close [_] - (release!))) - -(defn- add-suppressed! [error suppressed] - (when suppressed - (.addSuppressed ^Exception error ^Exception suppressed)) - error) +(defn- await-lifecycle [completion] + (let [interrupted? (volatile! false)] + (try + ;; Observe interruption only after lifecycle ownership is settled. + (loop [] + (let [[completed? value] (try + [true (p/join completion)] + (catch InterruptedException _ + (vreset! interrupted? true) + [false nil]))] + (if completed? value (recur)))) + (finally + (when @interrupted? (px/interrupt)))))) -(defn- close-resources! [pool object-store] - (let [pool-error (try - (.close ^GenericKeyedObjectPool pool) - nil - (catch Exception error - error)) - object-store-error (try - (ffi/close-object-store! object-store) - nil - (catch Exception error - error))] - (cond - pool-error (throw (add-suppressed! pool-error object-store-error)) - object-store-error (throw object-store-error)))) +(defn- destroy-database [pooled-database] + (await-lifecycle + (ffi/close-database! (.getObject ^PooledObject pooled-database)))) (defn- database-pool-factory [object-store] (proxy [BaseKeyedPooledObjectFactory] [] (create [database-id] - (ffi/open-database! object-store database-id)) + (await-lifecycle (ffi/open-database! object-store database-id))) (wrap [database] (DefaultPooledObject. database)) (validateObject [_ pooled-database] (ffi/usable-database? (.getObject ^PooledObject pooled-database))) - (destroyObject [_ pooled-database] - (ffi/close-database! (.getObject ^PooledObject pooled-database))))) - -(defn- database-pool [object-store max-open-databases] - (let [config (GenericKeyedObjectPoolConfig.)] - (.setMaxTotal config max-open-databases) - (.setMaxTotalPerKey config 1) - (.setMaxIdlePerKey config 1) - (.setTestOnReturn config true) - (GenericKeyedObjectPool. (database-pool-factory object-store) config))) + (destroyObject + ([_ pooled-database] (destroy-database pooled-database)) + ([_ pooled-database _mode] (destroy-database pooled-database))))) (defn open {:malli/schema [:-> @@ -65,42 +48,36 @@ [:fn #(instance? Connection %)]]} [{:keys [object-store-url max-open-databases revision-retention-days] :or {revision-retention-days 30}}] - (let [object-store (ffi/open-object-store object-store-url)] - (try - (let [pool (database-pool object-store max-open-databases)] - (connection - {:revision-retention-days revision-retention-days - :borrow-database (fn [database-id] - (.borrowObject ^GenericKeyedObjectPool pool database-id)) - :return-database (fn [database-id database] - (.returnObject ^GenericKeyedObjectPool pool - database-id - database)) - :close-resources (fn [] - (close-resources! pool object-store))})) - (catch Exception error - (let [cleanup-error (try - (ffi/close-object-store! object-store) - nil - (catch Exception caught - caught))] - (throw (add-suppressed! error cleanup-error))))))) + (let [object-store (ffi/open-object-store object-store-url) + config (doto (GenericKeyedObjectPoolConfig.) + (.setMaxTotal max-open-databases) + (.setMaxTotalPerKey 1) + (.setMaxIdlePerKey 1) + (.setTestOnBorrow true) + (.setTestOnReturn true))] + (->Connection (GenericKeyedObjectPool. (database-pool-factory object-store) config) + object-store revision-retention-days))) -(defn database - [{:keys [borrow-database return-database]} {:keys [database-id]}] - (let [raw-database (borrow-database database-id) - closed? (atom false)] - (->DatabaseLease - raw-database - #(when (compare-and-set! closed? false true) - (return-database database-id raw-database))))) +(defn ^:no-doc call-with-database [{:keys [pool]} {:keys [database-id f]}] + (let [database (.borrowObject ^GenericKeyedObjectPool pool database-id)] + (try + (when (px/interrupted? :current) + (throw (InterruptedException.))) + (f database) + (finally + (.returnObject ^GenericKeyedObjectPool pool database-id database))))) -(defn read-only-snapshot [{:keys [raw-database]}] - (ffi/await (ffi/open-snapshot raw-database))) +(defmacro with-database + [[binding connection options] & body] + `(call-with-database ~connection + (assoc ~options :f (fn [~binding] ~@body)))) -(defn writable-transaction [{:keys [raw-database]}] - (ffi/await (ffi/begin-transaction raw-database))) +(defn subscribe [connection {:keys [database-id]}] + (with-database [database connection {:database-id database-id}] + (ffi/subscribe database))) -(defn close [{:keys [close-resources]}] - (close-resources) +(defn close [{:keys [pool object-store]}] + (try + (.close ^GenericKeyedObjectPool pool) + (finally (ffi/close-object-store! object-store))) nil) diff --git a/src/linear/adapter/slatedb/ffi.clj b/src/linear/adapter/slatedb/ffi.clj index e7974f1..e2e3723 100644 --- a/src/linear/adapter/slatedb/ffi.clj +++ b/src/linear/adapter/slatedb/ffi.clj @@ -2,10 +2,9 @@ (:refer-clojure :exclude [await]) (:require [cognitect.anomalies :as anomaly] - [promesa.core :as p] - [promesa.exec :as px]) + [promesa.core :as p]) (:import - (io.slatedb.uniffi CloseReason Db DbBuilder DbSnapshot DbTransaction Error$Closed Error$Data Error$Internal Error$Invalid Error$Transaction Error$Unavailable IsolationLevel ObjectStore PutOptions Ttl$ExpireAtMillis Ttl$NoExpiry WriteHandle) + (io.slatedb.uniffi CloseReason Db DbBuilder DbSnapshot DbStatus DbStatusSubscription DbTransaction Error$Closed Error$Data Error$Internal Error$Invalid Error$Transaction Error$Unavailable IsolationLevel ObjectStore PutOptions Ttl$ExpireAtMillis Ttl$NoExpiry WriteHandle) (java.util.concurrent CompletionException ExecutionException))) (defn- closed-category [^Error$Closed error] @@ -54,7 +53,7 @@ (defn await [promise] (try - (px/await! promise) + (p/join promise) (catch Exception error (throw (unwrap-completion error))))) @@ -93,30 +92,24 @@ nil) (defn open-database! [object-store database-id] - (with-open [builder (DbBuilder. ^String database-id ^ObjectStore object-store)] - (await (invoke :open-database #(.build builder))))) + (invoke :open-database + (fn [] + (let [builder (DbBuilder. ^String database-id ^ObjectStore object-store)] + (p/finally (invoke :open-database #(.build builder)) + (fn [_value _error] (.close builder))))))) (defn close-database! [database] - (let [shutdown-error (try - (await (invoke :shutdown-database #(.shutdown ^Db database))) - nil - (catch Exception error - error)) - close-error (try - (.close ^Db database) - nil - (catch Exception error - error))] - (cond - shutdown-error - (do - (when close-error - (.addSuppressed ^Exception shutdown-error ^Exception close-error)) - (throw shutdown-error)) - - close-error - (throw close-error))) - nil) + (p/handle (invoke :shutdown-database #(.shutdown ^Db database)) + (fn [_value shutdown-error] + (let [close-error (try (.close ^Db database) nil + (catch Exception error error))] + (cond + shutdown-error + (do + (when close-error + (.addSuppressed ^Exception shutdown-error ^Exception close-error)) + (throw shutdown-error)) + close-error (throw close-error)))))) (defn usable-database? [database] (try @@ -127,6 +120,36 @@ (defn open-snapshot [database] (invoke :open-snapshot #(.snapshot ^Db database))) +(defn- await-change [raw-subscription durable-sequence timeout-ms] + (let [waiting (.waitForChange ^DbStatusSubscription raw-subscription + (long durable-sequence) (unchecked-int timeout-ms))] + (try + (when-let [^DbStatus status (await (invoke :await-change (constantly waiting)))] + (when-let [reason (.closeReason status)] + (when-not (identical? CloseReason/CLEAN reason) + (throw (classify-error (Error$Closed. reason "Database closed during revision wait") + {:operation :await-change}))))) + nil + (finally + (p/cancel waiting))))) + +(defn subscribe [database] + (try + (.subscribe ^Db database) + (catch Exception error + (throw (classify-error error {:operation :subscribe}))))) + +(defn ^:no-doc change-waiter [subscription] + (let [durable-sequence (-> ^DbStatusSubscription subscription .status .durableSeq)] + (fn [{:keys [timeout-ms]}] + (await-change subscription durable-sequence timeout-ms)))) + +(defmacro with-change-notifications + [[binding subscription] & body] + `(with-open [^DbStatusSubscription subscription# ~subscription] + (let [~binding (change-waiter subscription#)] + ~@body))) + (defn close-snapshot! [snapshot] (try (.close ^DbSnapshot snapshot) @@ -168,3 +191,17 @@ (catch Exception error (throw (classify-error error {:operation :close-transaction})))) nil) + +(defmacro with-snapshot + [[binding database] & body] + `(let [snapshot# (await (open-snapshot ~database))] + (try + (let [~binding snapshot#] ~@body) + (finally (close-snapshot! snapshot#))))) + +(defmacro with-transaction + [[binding database] & body] + `(let [transaction# (await (begin-transaction ~database))] + (try + (let [~binding transaction#] ~@body) + (finally (close-transaction! transaction#))))) diff --git a/src/linear/adapter/slatedb/store.clj b/src/linear/adapter/slatedb/store.clj index 9da731f..cf8efe9 100644 --- a/src/linear/adapter/slatedb/store.clj +++ b/src/linear/adapter/slatedb/store.clj @@ -7,10 +7,7 @@ [linear.adapter.slatedb.ffi :as ffi] [linear.adapter.slatedb.key :as key] [linear.adapter.slatedb.snapshot :as snapshot] - [linear.usecase.database.revisions :as revisions]) - (:import - (io.slatedb.uniffi DbTransaction) - (java.lang AutoCloseable))) + [linear.usecase.database.revisions :as revisions])) (defn- current-state-records [keychain revision head] (conj (mapv (fn [[page-id page]] @@ -63,56 +60,43 @@ (defn- publish! [store revision database initialize?] (let [database-id (:id database) keychain (get-in database [:vault :keychain])] - (with-open [^AutoCloseable leased-database - (connection/database store {:database-id database-id}) - ^DbTransaction transaction - (connection/writable-transaction leased-database)] - (try - (let [read-values #(ffi/await - (ffi/read-transaction-values transaction %)) - head (if initialize? - (let [[value] (read-values [(key/head)])] - (when value - (throw (revision-conflict "Database HEAD already exists" {} nil))) - nil) - (snapshot/head-metadata read-values)) - expires-at (max (+ (System/currentTimeMillis) - (* (:revision-retention-days store) 86400000)) - (or (:history-expires-at head) 0)) - next-head (assoc (select-keys revision [:revision-id :database-page-count]) - :history-expires-at expires-at)] - (ensure-current-parent! (:revision-id head) revision) - (let [record-key (key/revision (:revision-id revision)) - undo {:parent head - :undo-pages (undo-pages read-values keychain head (vec (keys (:pages revision))))}] + (connection/with-database [database-handle store {:database-id database-id}] + (ffi/with-transaction [transaction database-handle] + (try + (let [read-values #(ffi/await + (ffi/read-transaction-values transaction %)) + head (if initialize? + (let [[value] (read-values [(key/head)])] + (when value + (throw (revision-conflict "Database HEAD already exists" {} nil))) + nil) + (snapshot/head-metadata read-values)) + expires-at (max (+ (System/currentTimeMillis) + (* (:revision-retention-days store) 86400000)) + (or (:history-expires-at head) 0)) + next-head (assoc (select-keys revision [:revision-id :database-page-count]) + :history-expires-at expires-at)] + (ensure-current-parent! (:revision-id head) revision) + (let [record-key (key/revision (:revision-id revision)) + undo {:parent head + :undo-pages (undo-pages read-values keychain head (vec (keys (:pages revision))))}] + (ffi/await + (ffi/write-expiring-values transaction + {:entries [[record-key (codec/encode-revision keychain {:record-key record-key :revision undo})]] + :expires-at expires-at}))) (ffi/await - (ffi/write-expiring-values transaction - {:entries [[record-key (codec/encode-revision keychain {:record-key record-key :revision undo})]] - :expires-at expires-at}))) - (ffi/await - (ffi/write-values transaction - (current-state-records keychain revision next-head))) - (ffi/await (ffi/commit-transaction transaction)) - revision) - (catch Exception error - (try - (ffi/await (ffi/rollback-transaction transaction)) - (catch Exception rollback-error - (add-suppressed! error rollback-error))) - (if (commit-conflict? error) - (throw (revision-conflict "SlateDB publish conflict" {} error)) - (throw error))))))) - -(defn- with-consistent-view [store database f] - (with-open [^AutoCloseable leased-database - (connection/database store {:database-id (:id database)})] - (let [raw-snapshot (connection/read-only-snapshot leased-database)] - (try - (let [read-values #(ffi/await (ffi/read-snapshot-values raw-snapshot %)) - view (snapshot/consistent-view {:read-values read-values :keychain (get-in database [:vault :keychain])})] - (f view)) - (finally - (ffi/close-snapshot! raw-snapshot)))))) + (ffi/write-values transaction + (current-state-records keychain revision next-head))) + (ffi/await (ffi/commit-transaction transaction)) + revision) + (catch Exception error + (try + (ffi/await (ffi/rollback-transaction transaction)) + (catch Exception rollback-error + (add-suppressed! error rollback-error))) + (if (commit-conflict? error) + (throw (revision-conflict "SlateDB publish conflict" {} error)) + (throw error)))))))) (extend-type linear.adapter.slatedb.connection.Connection revisions/Initializable @@ -121,11 +105,25 @@ revisions/ConsistentReadable (-read-consistently [store f database] - (with-consistent-view store database f)) + (connection/with-database [database-handle store {:database-id (:id database)}] + (ffi/with-snapshot [raw-snapshot database-handle] + (let [read-values #(ffi/await (ffi/read-snapshot-values raw-snapshot %)) + view (snapshot/consistent-view {:read-values read-values :keychain (get-in database [:vault :keychain])})] + (f view))))) revisions/RevisionWritable (-publish-next! [store revision database] - (publish! store revision database false))) + (publish! store revision database false)) + + revisions/RevisionAwaitable + (-await-revision [store {:keys [database-id revision-id timeout-ms]}] + (ffi/with-change-notifications [await-change (connection/subscribe store {:database-id database-id})] + (let [current-revision-id + (connection/with-database [database store {:database-id database-id}] + (ffi/with-snapshot [raw-snapshot database] + (:revision-id (snapshot/head-metadata #(ffi/await (ffi/read-snapshot-values raw-snapshot %))))))] + (when (= current-revision-id revision-id) + (await-change {:timeout-ms timeout-ms})))))) (defmethod integrant/init-key :linear.adapter.slatedb.store/store [_ options] diff --git a/src/linear/adapter/sqlite/connection.clj b/src/linear/adapter/sqlite/connection.clj index 42af94e..362d636 100644 --- a/src/linear/adapter/sqlite/connection.clj +++ b/src/linear/adapter/sqlite/connection.clj @@ -288,45 +288,47 @@ (sqlite-diagnostics database result))))) (catch Exception error error))) -(defn- with-statement [connection {:keys [sql parameters]} f] - (ensure-open! connection) - (let [database (:handle connection)] - (with-open [arena (mem/confined-arena)] - (let [statement (prepare! database arena sql) - finalization-error (volatile! nil) - outcome (try - (bind-parameters! database statement arena parameters) - {:value (f database statement)} - (catch Exception error {:error error}) - (finally - (vreset! finalization-error - (statement-finalization-error database statement))))] - (cond - (:error outcome) (throw (add-suppressed! (:error outcome) @finalization-error)) - @finalization-error (throw @finalization-error) - :else (:value outcome)))))) +(defmacro ^:private with-statement + [[binding connection options] & body] + `(let [connection# ~connection + options# ~options] + (ensure-open! connection#) + (let [database# (:handle connection#)] + (with-open [arena# (mem/confined-arena)] + (let [statement# (prepare! database# arena# (:sql options#)) + finalization-error# (volatile! nil) + outcome# (try + (bind-parameters! database# statement# arena# (:parameters options#)) + {:value (let [~binding statement#] ~@body)} + (catch Exception error# {:error error#}) + (finally + (vreset! finalization-error# + (statement-finalization-error database# statement#))))] + (cond + (:error outcome#) (throw (add-suppressed! (:error outcome#) @finalization-error#)) + @finalization-error# (throw @finalization-error#) + :else (:value outcome#))))))) (defn execute-statement {:malli/schema [:-> ::connection ::evaluator/statement :nil]} [connection arg-map] - (with-statement connection arg-map step-to-completion!)) + (with-statement [statement connection arg-map] + (step-to-completion! (:handle connection) statement))) (defn query-integers {:malli/schema [:-> ::connection ::evaluator/statement [:maybe [:vector :int]]]} [connection arg-map] - (with-statement - connection arg-map - (fn [database statement] - (let [result (ffi/step statement)] - (cond - (= ffi/sqlite-done result) nil - (= ffi/sqlite-row result) - (mapv (fn [index] - (when-not (= 1 (ffi/column-type statement {:index index})) - (fault "SQLite query returned a non-integer column" {:index index})) - (ffi/column-int64 statement {:index index})) - (range (ffi/column-count statement))) - :else (fault "SQLite query failed" (sqlite-diagnostics database result))))))) + (with-statement [statement connection arg-map] + (let [result (ffi/step statement)] + (cond + (= ffi/sqlite-done result) nil + (= ffi/sqlite-row result) + (mapv (fn [index] + (when-not (= 1 (ffi/column-type statement {:index index})) + (fault "SQLite query returned a non-integer column" {:index index})) + (ffi/column-int64 statement {:index index})) + (range (ffi/column-count statement))) + :else (fault "SQLite query failed" (sqlite-diagnostics (:handle connection) result)))))) (defn close {:malli/schema [:-> ::connection :nil]} diff --git a/src/linear/adapter/sqlite/evaluator.clj b/src/linear/adapter/sqlite/evaluator.clj index 0156af6..a725bae 100644 --- a/src/linear/adapter/sqlite/evaluator.clj +++ b/src/linear/adapter/sqlite/evaluator.clj @@ -2,7 +2,6 @@ (:require [clojure.java.io :as io] [cognitect.anomalies :as anomaly] - [diehard.core :refer [with-timeout]] [hden.ulid :refer [ulid]] [integrant.core :as integrant] [linear.adapter.sqlite.connection :as connection] @@ -13,9 +12,7 @@ [linear.usecase.database.evaluator :as evaluator] [linear.usecase.database.revisions :as revisions] [linear.usecase.healthcheck :as healthcheck] - [promesa.core :as p]) - (:import - (dev.failsafe TimeoutExceededException))) + [promesa.core :as p])) (def ^:private ^:const default-shutdown-timeout-ms 10000) @@ -85,26 +82,30 @@ (defn- end-evaluation! [state] (signal-if-drained! (swap! state update :active dec))) -(defn- with-snapshot [evaluator snapshot f] - (begin-evaluation! (:state evaluator)) - (try - (let [path (str "/linear/" (ulid) ".db") - filesystem (evaluation/evaluation snapshot {:path path}) - invocation (vfs/mount (:resources evaluator) {:path path :filesystem filesystem})] - (try - (let [database (call-sqlite invocation - {:operation :open :path path} - #(connection/open {:path path - :vfs-name (:name (:resources evaluator))}))] - (try - (configure! invocation database) - (f invocation database filesystem) - (finally - (call-sqlite invocation {:operation :close} #(connection/close database))))) - (finally - (vfs/unmount invocation)))) - (finally - (end-evaluation! (:state evaluator))))) +(defmacro ^:private with-evaluation + [[binding evaluator options] & body] + `(let [evaluator# ~evaluator + snapshot# (:snapshot ~options)] + (begin-evaluation! (:state evaluator#)) + (try + (let [path# (str "/linear/" (ulid) ".db") + filesystem# (evaluation/evaluation snapshot# {:path path#}) + invocation# (vfs/mount (:resources evaluator#) {:path path# :filesystem filesystem#})] + (try + (let [database# (call-sqlite invocation# + {:operation :open :path path#} + #(connection/open {:path path# + :vfs-name (:name (:resources evaluator#))}))] + (try + (configure! invocation# database#) + (let [~binding {:invocation invocation# :database database# :filesystem filesystem#}] + ~@body) + (finally + (call-sqlite invocation# {:operation :close} #(connection/close database#))))) + (finally + (vfs/unmount invocation#)))) + (finally + (end-evaluation! (:state evaluator#)))))) (defn- read-progress [invocation database client-id] (let [[table-count] (call-sqlite invocation {:operation :read-sync-progress} @@ -135,13 +136,13 @@ (vfs/ok? resources))) evaluator/SyncProgressReadable (-sync-progress [this {:keys [snapshot client-id]}] - (with-snapshot this snapshot - (fn [invocation database _] + (with-evaluation [evaluation this {:snapshot snapshot}] + (let [{:keys [invocation database]} evaluation] (read-progress invocation database client-id)))) evaluator/Evaluator (-evaluate [this {:keys [snapshot command]}] - (with-snapshot this snapshot - (fn [invocation database filesystem] + (with-evaluation [evaluation this {:snapshot snapshot}] + (let [{:keys [invocation database filesystem]} evaluation] (execute-command! invocation database command) (when-let [delta (evaluation/commit filesystem)] (revision snapshot delta)))))) @@ -163,13 +164,7 @@ (defmethod integrant/halt-key! :linear.adapter.sqlite.evaluator/evaluator [_ evaluator] (signal-if-drained! (swap! (:state evaluator) assoc :status :draining)) - (let [drained? (try - (with-timeout {:timeout-ms (:shutdown-timeout-ms evaluator) - :interrupt? true} - (p/join (:drained @(:state evaluator)))) - true - (catch TimeoutExceededException _ - false))] + (let [drained? (p/await (:drained @(:state evaluator)) (:shutdown-timeout-ms evaluator) false)] (if drained? (do (vfs/drain! (:resources evaluator)) diff --git a/src/linear/handler/turso/protobuf.clj b/src/linear/handler/turso/protobuf.clj index 1972e15..4ff9e49 100644 --- a/src/linear/handler/turso/protobuf.clj +++ b/src/linear/handler/turso/protobuf.clj @@ -84,6 +84,8 @@ (malformed "Pull client revision has an invalid wire type")) 4 (if (= 0 wire-type) (let [[value end] (read-varint data next-offset)] + (when-not (<= 0 value 0xffffffff) + (malformed "Pull long poll timeout must be a uint32")) (recur (long end) (assoc pull :long-poll-timeout-ms value))) (malformed "Pull long poll timeout has an invalid wire type")) 5 (if (= 2 wire-type) diff --git a/src/linear/handler/turso/pull.clj b/src/linear/handler/turso/pull.clj index 93838cc..171f706 100644 --- a/src/linear/handler/turso/pull.clj +++ b/src/linear/handler/turso/pull.clj @@ -31,7 +31,6 @@ (defn- validate-options [pull] (when (or (not (zero? (get pull :encoding 0))) (not (zero? (get pull :stream-kind 0))) - (not (zero? (get pull :long-poll-timeout-ms 0))) (seq (:server-query-selector pull)) (seq (:client-pages pull))) (throw (ex-info "Pull option is not supported" @@ -45,6 +44,7 @@ _ (validate-options pull) result (database/pull context {:server-revision (not-empty (:server-revision pull)) :client-revision (not-empty (:client-revision pull)) + :long-poll-timeout-ms (get pull :long-poll-timeout-ms 0) :page-ids (protobuf/decode-page-selector (:server-pages-selector pull)) :actor (:sub identity) diff --git a/src/linear/server/jetty.clj b/src/linear/server/jetty.clj index abe551c..900597f 100644 --- a/src/linear/server/jetty.clj +++ b/src/linear/server/jetty.clj @@ -3,7 +3,11 @@ [integrant.core :as ig]) (:import (org.eclipse.jetty.http UriCompliance UriCompliance$Violation) - (org.eclipse.jetty.server Connector HttpConnectionFactory Server))) + (org.eclipse.jetty.server Connector HttpConnectionFactory Server) + (org.eclipse.jetty.util.thread VirtualThreadPool))) + +(defmethod ig/init-key ::thread-pool [_ _] + (VirtualThreadPool.)) (defn- allow-encoded-path-separators! [^Server server] (let [uri-compliance diff --git a/src/linear/usecase/core.clj b/src/linear/usecase/core.clj index 82fc53e..178a7e3 100644 --- a/src/linear/usecase/core.clj +++ b/src/linear/usecase/core.clj @@ -28,5 +28,14 @@ [{::keys [revision-store]}] revision-store) +(defn revision-awaitable + [{::keys [revision-store]}] + revision-store) + +(defn max-long-poll-timeout-ms + {:malli/schema [:-> :map pos-int?]} + [{::keys [max-long-poll-timeout-ms]}] + max-long-poll-timeout-ms) + (defn checkables [context] [(transactable context) (evaluator context)]) diff --git a/src/linear/usecase/database.clj b/src/linear/usecase/database.clj index e56a5bf..562a5c8 100644 --- a/src/linear/usecase/database.clj +++ b/src/linear/usecase/database.clj @@ -191,7 +191,7 @@ error)) (throw error))))) -(defn pull +(defn- pull-once [context {:keys [actor database-id] :as pull-options}] (with-database [database context {:database-id database-id :actor actor @@ -199,6 +199,23 @@ :read-only true}] (model/pull database pull-options))) +(defn pull + [context {:keys [database-id client-revision server-revision long-poll-timeout-ms] + :or {long-poll-timeout-ms 0} + :as pull-options}] + (let [result (pull-once context pull-options)] + (if (and client-revision (nil? server-revision) + (pos? long-poll-timeout-ms) + (= client-revision (:server-revision result))) + (do + (revisions/await-revision (core/revision-awaitable context) + {:database-id database-id + :revision-id client-revision + :timeout-ms (min long-poll-timeout-ms + (core/max-long-poll-timeout-ms context))}) + (pull-once context pull-options)) + result))) + (defn sync-progress [context {:keys [actor database-id] :as arg-map}] (with-database [database context {:database-id database-id diff --git a/src/linear/usecase/database/revisions.clj b/src/linear/usecase/database/revisions.clj index e1930a9..541fadd 100644 --- a/src/linear/usecase/database/revisions.clj +++ b/src/linear/usecase/database/revisions.clj @@ -91,6 +91,19 @@ (defprotocol RevisionWritable (-publish-next! [writer revision database])) +(defprotocol RevisionAwaitable + (-await-revision [reader arg-map])) + +(defn await-revision + {:malli/schema [:-> [:fn #(satisfies? RevisionAwaitable %)] + [:map + [:database-id :string] + [:revision-id ::revision-id] + [:timeout-ms pos-int?]] + :nil]} + [reader arg-map] + (-await-revision reader arg-map)) + (defprotocol Initializable (-initialize! [writer arg-map])) diff --git a/test/e2e/turso-sync.mjs b/test/e2e/turso-sync.mjs index 766364f..d8e35f2 100644 --- a/test/e2e/turso-sync.mjs +++ b/test/e2e/turso-sync.mjs @@ -109,6 +109,75 @@ try { await readerClient.exec("UPDATE t SET value = 'reader-local' WHERE id = 1"); await assert.rejects(() => readerClient.push(), /403/); + const shortPollClient = await connectClient("long-poll-short", { longPollTimeoutMs: 100 }); + const shortPollStart = performance.now(); + assert.equal(await shortPollClient.pull(), false); + const shortPollElapsed = performance.now() - shortPollStart; + assert(shortPollElapsed >= 90 && shortPollElapsed < 4000, + `requested short wait was not respected: ${shortPollElapsed} ms`); + + // The official SDK must accept a normal empty pull before its requested + // timeout, then keep pulling from the same revision. + const cappedClient = await connectClient("long-poll-capped", { + longPollTimeoutMs: 0xffffffff + }); + for (let attempt = 0; attempt < 2; attempt += 1) { + const start = performance.now(); + assert.equal(await cappedClient.pull(), false); + const elapsed = performance.now() - start; + assert(elapsed >= 4900, `server cap returned too early: ${elapsed} ms`); + assert(elapsed < 9000, `server cap did not bound waiting: ${elapsed} ms`); + } + + let observePull; + let abortPull; + const longPollClient = await connectClient("long-poll", { + longPollTimeoutMs: 5000, + fetch: async (input, init) => { + if (String(input).endsWith("/pull-updates") && observePull) { + const observed = observePull; + observePull = undefined; + observed(); + if (abortPull) { + return timedFetch(input, { ...init, signal: abortPull.signal }); + } + } + return timedFetch(input, init); + } + }); + const waiting = new Promise(resolve => { observePull = resolve; }); + let completed = false; + const pulling = longPollClient.pull().then(changed => { + completed = true; + return changed; + }); + await waiting; + await new Promise(resolve => setTimeout(resolve, 100)); + assert.equal(completed, false); + const wakeStart = performance.now(); + await clientA.exec("UPDATE t SET value = 'long-poll-update' WHERE id = 1"); + await clientA.push(); + assert.equal(await pulling, true); + assert(performance.now() - wakeStart < 4000, "published update did not wake the pull before its deadline"); + assert.deepEqual(await rows(longPollClient, "SELECT value FROM t WHERE id = 1"), [ + { value: "long-poll-update" } + ]); + + const disconnecting = new Promise(resolve => { observePull = resolve; }); + abortPull = new AbortController(); + const interrupted = assert.rejects(() => longPollClient.pull()); + await disconnecting; + abortPull.abort(); + await interrupted; + abortPull = undefined; + await clientA.exec("UPDATE t SET value = 'from-js' WHERE id = 1"); + await clientA.push(); + assert.equal(await longPollClient.pull(), true); + assert.deepEqual(await rows(longPollClient, "SELECT value FROM t WHERE id = 1"), [ + { value: "from-js" } + ]); + await clientB.pull(); + await assert.rejects( () => connectClient("ungranted", { authToken: ungrantedToken }), /403/ diff --git a/test/linear/adapter/slatedb/connection_test.clj b/test/linear/adapter/slatedb/connection_test.clj index 1c4f999..929182a 100644 --- a/test/linear/adapter/slatedb/connection_test.clj +++ b/test/linear/adapter/slatedb/connection_test.clj @@ -1,80 +1,305 @@ (ns linear.adapter.slatedb.connection-test (:require [clojure.test :refer [deftest is]] - [cognitect.anomalies :as anomaly] [linear.adapter.slatedb.connection :as connection] - [linear.adapter.slatedb.ffi :as ffi]) - (:import - (java.lang AutoCloseable))) - -(deftest database-lease-ends-after-the-scoped-operation - (let [events (atom []) - connection (connection/connection - {:borrow-database (fn [database-id] - (swap! events conj [:borrow database-id]) - ::database) - :return-database (fn [database-id database] - (swap! events conj [:return database-id database])) - :close-resources (fn [])})] - (with-open [_database (connection/database connection {:database-id "d-1"})] - (swap! events conj :call)) - (is (= [[:borrow "d-1"] - :call - [:return "d-1" ::database]] - @events)))) - -(deftest database-is-returned-after-a-fenced-operation - (let [events (atom []) - failure (ex-info "fenced" - {::anomaly/category ::anomaly/unavailable - :reason ::ffi/fenced}) - connection (connection/connection - {:borrow-database (constantly ::database) - :return-database (fn [database-id database] - (swap! events conj [:return database-id database])) - :close-resources (fn [])}) - caught (try - (with-open [_database (connection/database connection {:database-id "d-1"})] - (throw failure)) - (catch Exception error - error))] - (is (identical? failure caught)) - (is (= [[:return "d-1" ::database]] @events)))) - -(deftest return-failure-is-propagated-on-close - (let [events (atom []) - return-error (ex-info "return failed" {}) - connection (connection/connection - {:borrow-database (constantly ::database) - :return-database (fn [_ _] - (swap! events conj :return) - (throw return-error)) - :close-resources (fn [])}) - caught (try - (with-open [_database (connection/database connection {:database-id "d-1"})] - :done) - (catch Exception error - error))] - (is (identical? return-error caught)) - (is (= [:return] @events)))) - -(deftest database-lease-is-returned-only-once - (let [returns (atom 0) - connection (connection/connection - {:borrow-database (constantly ::database) - :return-database (fn [_ _] - (swap! returns inc)) - :close-resources (fn [])}) - database (connection/database connection {:database-id "d-1"})] - (.close ^AutoCloseable database) - (.close ^AutoCloseable database) - (is (= 1 @returns)))) - -(deftest close-releases-owned-resources - (let [closed? (atom false) - connection (connection/connection - {:borrow-database (fn [_]) - :return-database (fn [_ _]) - :close-resources #(reset! closed? true)})] - (is (nil? (connection/close connection))) - (is @closed?))) + [linear.adapter.slatedb.ffi :as ffi] + [promesa.core :as p] + [promesa.exec :as px])) + +(defn- new-connection [] + (connection/open {:object-store-url "memory:///" :max-open-databases 1})) + +(defn- use-database [connection id] + (connection/with-database [database connection {:database-id id}] + (ffi/usable-database? database))) + +(deftest ^:integration reader-waits-until-the-writer-returns-the-database + (let [connection (new-connection) + ready (p/deferred) + finish (p/deferred) + writer (p/thread + (connection/with-database [database connection {:database-id "a"}] + (ffi/with-transaction [transaction database] + (try + (ffi/await (ffi/write-values transaction [[(byte-array [1]) (byte-array [2])]])) + (p/resolve ready true) + (p/join finish) + (finally (ffi/await (ffi/rollback-transaction transaction)))))))] + (try + (is (= true (p/await ready 5000 ::timeout))) + (let [reader (p/thread + (connection/with-database [database connection {:database-id "a"}] + (ffi/with-snapshot [snapshot database] + (ffi/await (ffi/read-snapshot-values snapshot [(byte-array [1])])))))] + (try (is (= ::waiting (p/await reader 100 ::waiting))) + (p/resolve finish true) + (is (= [nil] (p/await reader 5000 ::timeout))) + (finally (p/resolve finish true) (p/await reader 15000 ::timeout)))) + (finally + (p/resolve finish true) + (p/await writer 15000 ::timeout) + (connection/close connection))))) + +(deftest ^:integration writers-serialize-and-interruption-releases-their-use + (let [connection (new-connection) + ready (p/deferred) + finish (p/deferred) + writer (p/thread + (connection/with-database [_ connection {:database-id "a"}] + (p/resolve ready true) + (p/join finish))) + attempting (p/deferred) + entered (p/deferred) + interrupted (p/deferred)] + (try + (is (= true (p/await ready 5000 ::timeout))) + (let [request-thread (p/deferred) + next-writer (p/thread + (p/resolve request-thread (px/current-thread)) + (p/resolve attempting true) + (try + (connection/with-database [_ connection {:database-id "a"}] + (p/resolve entered true)) + (catch InterruptedException _ (p/resolve interrupted true))))] + (try + (is (= true (p/await attempting 5000 ::timeout))) + (is (= ::waiting (p/await entered 100 ::waiting))) + (px/interrupt (p/join request-thread)) + (is (= true (p/await interrupted 5000 ::timeout))) + (finally (p/await next-writer 15000 ::timeout)))) + (p/resolve finish true) + (is (= true (p/await writer 5000 ::timeout))) + (is (use-database connection "b")) + (finally + (p/resolve finish true) + (p/await writer 15000 ::timeout) + (connection/close connection))))) + +(deftest ^:integration active-snapshot-survives-capacity-pressure + (let [connection (new-connection) + ready (p/deferred) + finish (p/deferred) + reader (p/thread + (connection/with-database [database connection {:database-id "a"}] + (ffi/with-snapshot [snapshot database] + (p/resolve ready true) + (p/join finish) + (ffi/await (ffi/read-snapshot-values snapshot [(byte-array [1])]))))) + attempting (p/deferred)] + (try + (is (= true (p/await ready 5000 ::timeout))) + (let [other (p/thread (p/resolve attempting true) (use-database connection "b"))] + (try + (is (= true (p/await attempting 5000 ::timeout))) + (is (= ::waiting (p/await other 100 ::waiting))) + (p/resolve finish true) + (is (= [nil] (p/await reader 5000 ::timeout))) + (is (= true (p/await other 5000 ::timeout))) + (finally (p/resolve finish true) (p/await other 15000 ::timeout)))) + (finally + (p/resolve finish true) + (p/await reader 15000 ::timeout) + (connection/close connection))))) + +(deftest ^:integration subscription-wakes-on-eviction-without-holding-capacity + (let [connection (new-connection)] + (try + (ffi/with-change-notifications [await-change (connection/subscribe connection {:database-id "a"})] + (is (use-database connection "b")) + (let [waiting (p/thread (await-change {:timeout-ms 10000}))] + (try + (is (nil? (p/await waiting 1000 ::timeout))) + (is (use-database connection "a")) + (finally (p/await waiting 15000 ::timeout))))) + (finally (connection/close connection))))) + +(deftest ^:integration invalid-handles-and-callback-errors-release-capacity + (let [connection (new-connection)] + (try + (is (thrown-with-msg? Exception #"callback" + (connection/with-database [_ connection {:database-id "a"}] + (throw (ex-info "callback" {}))))) + (connection/with-database [database connection {:database-id "a"}] + (ffi/await (ffi/close-database! database))) + (is (use-database connection "a")) + (is (use-database connection "b")) + (finally (connection/close connection))))) + +(deftest ^:integration close-rejects-new-borrows-without-waiting-for-active-borrow + (let [connection (new-connection)] + (try + (connection/with-database [database connection {:database-id "a"}] + (is (nil? (connection/close connection))) + (is (ffi/usable-database? database)) + (is (thrown? IllegalStateException (use-database connection "b")))) + (is (thrown? IllegalStateException (use-database connection "a"))) + (finally (connection/close connection))))) + +(deftest ^:integration concurrent-first-access-serializes-all-borrows + (let [connection (new-connection) + start (p/deferred) + active (atom 0) + readers (mapv (fn [_] + (p/thread + (p/join start) + (connection/with-database [database connection {:database-id "a"}] + (try + (is (= 1 (swap! active inc))) + (ffi/usable-database? database) + (finally (swap! active dec)))))) + (range 8))] + (try + (p/resolve start true) + (doseq [reader readers] (is (= true (p/await reader 5000 ::timeout)))) + (finally + (p/resolve start true) + (doseq [reader readers] (p/await reader 15000 ::timeout)) + (connection/close connection))))) + +(deftest ^:integration interrupted-capacity-wait-does-not-block-later-acquisition + (let [connection (new-connection) + ready (p/deferred) + finish (p/deferred) + interrupted (p/deferred) + reader (p/thread + (connection/with-database [_ connection {:database-id "a"}] + (p/resolve ready true) + (p/join finish)))] + (try + (is (= true (p/await ready 5000 ::timeout))) + (let [request-thread (p/deferred) + waiter (p/thread + (p/resolve request-thread (px/current-thread)) + (try (use-database connection "b") + (catch InterruptedException _ (p/resolve interrupted true))))] + (try + (is (= ::waiting (p/await waiter 100 ::waiting))) + (px/interrupt (p/join request-thread)) + (is (= true (p/await interrupted 5000 ::timeout))) + (p/resolve finish true) + (is (= true (p/await reader 5000 ::timeout))) + (is (use-database connection "b")) + (finally (p/await waiter 15000 ::timeout)))) + (finally + (p/resolve finish true) + (p/await reader 15000 ::timeout) + (connection/close connection))))) + +(deftest ^:integration fenced-handle-is-retired-before-the-next-acquisition + (let [connection (new-connection)] + (try + (connection/with-database [database connection {:database-id "a"}] + (let [replacement (ffi/await (ffi/open-database! (:object-store connection) "a"))] + (try + (let [error (try + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(byte-array [1]) (byte-array [2])]])) + (ffi/await (ffi/commit-transaction transaction))) + nil + (catch Exception error error))] + (is (= ::ffi/fenced (:reason (ex-data error))))) + (finally (ffi/await (ffi/close-database! replacement)))))) + (is (use-database connection "a")) + (is (use-database connection "b")) + (finally (connection/close connection))))) + +(deftest ^:integration failed-open-does-not-leave-a-capacity-reservation + (let [connection (new-connection)] + (try + (is (thrown? Exception (use-database connection nil))) + (is (use-database connection "a")) + (is (thrown? Exception (use-database connection nil))) + (is (use-database connection "b")) + (finally (connection/close connection))))) + +(deftest ^:integration interrupted-first-acquisition-does-not-cancel-lifecycle-work + (let [connection (new-connection) + request (p/thread + (px/interrupt) + (try + (connection/with-database [_ connection {:database-id "a"}] + true) + (catch InterruptedException _ ::interrupted)))] + (try + (is (= ::interrupted (p/await request 5000 ::timeout))) + (is (use-database connection "a")) + (is (use-database connection "b")) + (finally + (p/await request 15000 ::timeout) + (connection/close connection))))) + +(deftest ^:integration interrupted-writer-does-not-let-its-successor-overtake + (let [connection (new-connection) + ready (p/deferred) + finish (p/deferred) + first-writer (p/thread + (connection/with-database [_ connection {:database-id "a"}] + (p/resolve ready true) + (p/join finish))) + attempting (p/deferred) + interrupted (p/deferred) + entered (p/deferred)] + (try + (is (= true (p/await ready 5000 ::timeout))) + (let [request-thread (p/deferred) + middle-writer (p/thread + (p/resolve request-thread (px/current-thread)) + (p/resolve attempting true) + (try + (connection/with-database [_ connection {:database-id "a"}] + (throw (AssertionError. "Interrupted writer entered its scope"))) + (catch InterruptedException _ (p/resolve interrupted true))))] + (try + (is (= true (p/await attempting 5000 ::timeout))) + (is (= ::waiting (p/await middle-writer 100 ::waiting))) + (px/interrupt (p/join request-thread)) + (is (= true (p/await interrupted 5000 ::timeout))) + (let [last-writer (p/thread + (connection/with-database [_ connection {:database-id "a"}] + (p/resolve entered true)))] + (try + (is (= ::waiting (p/await entered 100 ::waiting))) + (p/resolve finish true) + (is (= true (p/await first-writer 5000 ::timeout))) + (is (= true (p/await entered 5000 ::timeout))) + (p/await last-writer 5000 ::timeout) + (finally (p/resolve finish true) (p/await last-writer 15000 ::timeout)))) + (finally (p/await middle-writer 15000 ::timeout)))) + (finally + (p/resolve finish true) + (try (p/await first-writer 5000 ::timeout) + (finally + (p/await first-writer 15000 ::timeout) + (connection/close connection))))))) + +(deftest ^:integration competing-databases-progress-with-a-single-open-slot + (let [connection (new-connection) + start (p/deferred) + readers (mapv (fn [id] + (p/thread + (p/join start) + (vec (repeatedly 3 #(use-database connection id))))) + ["a" "a" "b" "b"])] + (try + (p/resolve start true) + (doseq [reader readers] + (is (= [true true true] (p/await reader 15000 ::timeout)))) + (finally + (p/resolve start true) + (try + (doseq [reader readers] (p/await reader 15000 ::timeout)) + (finally (connection/close connection))))))) + +(deftest ^:integration subscription-observes-publish-before-waiting + (let [connection (new-connection)] + (try + (ffi/with-change-notifications [await-change (connection/subscribe connection {:database-id "a"})] + (connection/with-database [database connection {:database-id "a"}] + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(byte-array [1]) (byte-array [2])]])) + (ffi/await (ffi/commit-transaction transaction)))) + (let [waiting (p/thread (await-change {:timeout-ms 10000}))] + (try + (is (nil? (p/await waiting 1000 ::timeout))) + (finally (p/await waiting 15000 ::timeout))))) + (finally (connection/close connection))))) diff --git a/test/linear/adapter/slatedb/ffi_test.clj b/test/linear/adapter/slatedb/ffi_test.clj index 234e59a..59a5820 100644 --- a/test/linear/adapter/slatedb/ffi_test.clj +++ b/test/linear/adapter/slatedb/ffi_test.clj @@ -51,17 +51,14 @@ (deftest ^:integration reads-and-writes-value-collections (let [object-store (ffi/open-object-store "memory:///") - database (ffi/open-database! object-store "d-ffi-test")] + database (ffi/await (ffi/open-database! object-store "d-ffi-test"))] (try - (let [transaction (ffi/await (ffi/begin-transaction database))] - (try - (ffi/await (ffi/write-values transaction - [[(utf8 "a") (utf8 "one")] - [(utf8 "b") (utf8 "two")]])) - (ffi/await (ffi/commit-transaction transaction)) - (finally - (ffi/close-transaction! transaction)))) - (let [transaction (ffi/await (ffi/begin-transaction database))] + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction + [[(utf8 "a") (utf8 "one")] + [(utf8 "b") (utf8 "two")]])) + (ffi/await (ffi/commit-transaction transaction))) + (ffi/with-transaction [transaction database] (try (let [values (ffi/await (ffi/read-transaction-values transaction @@ -73,39 +70,58 @@ (is (nil? (nth values 1))) (is (Arrays/equals (utf8 "one") (nth values 2)))) (finally - (ffi/await (ffi/rollback-transaction transaction)) - (ffi/close-transaction! transaction)))) + (ffi/await (ffi/rollback-transaction transaction))))) (finally - (ffi/close-database! database) + (ffi/await (ffi/close-database! database)) (ffi/close-object-store! object-store))))) (deftest ^:integration database-snapshot-retains-its-read-view (let [object-store (ffi/open-object-store "memory:///") - database (ffi/open-database! object-store "d-snapshot-test")] + database (ffi/await (ffi/open-database! object-store "d-snapshot-test"))] (try - (let [transaction (ffi/await (ffi/begin-transaction database))] - (try - (ffi/await (ffi/write-values transaction [[(utf8 "a") (utf8 "one")]])) - (ffi/await (ffi/commit-transaction transaction)) - (finally - (ffi/close-transaction! transaction)))) - (let [snapshot (ffi/await (ffi/open-snapshot database))] - (try - (let [transaction (ffi/await (ffi/begin-transaction database))] - (try - (ffi/await (ffi/write-values transaction [[(utf8 "a") (utf8 "two")]])) - (ffi/await (ffi/commit-transaction transaction)) - (finally - (ffi/close-transaction! transaction)))) - (let [[value] (ffi/await - (ffi/read-snapshot-values snapshot [(utf8 "a")]))] - (is (Arrays/equals (utf8 "one") value))) - (finally - (ffi/close-snapshot! snapshot)))) + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(utf8 "a") (utf8 "one")]])) + (ffi/await (ffi/commit-transaction transaction))) + (ffi/with-snapshot [snapshot database] + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(utf8 "a") (utf8 "two")]])) + (ffi/await (ffi/commit-transaction transaction))) + (let [[value] (ffi/await + (ffi/read-snapshot-values snapshot [(utf8 "a")]))] + (is (Arrays/equals (utf8 "one") value)))) (finally - (ffi/close-database! database) + (ffi/await (ffi/close-database! database)) (ffi/close-object-store! object-store))))) +(deftest ^:integration interruption-exits-change-notifications-scope + (with-open [object-store (ffi/open-object-store "memory:///")] + (let [database (ffi/await (ffi/open-database! object-store "d-subscription-test")) + ready (p/deferred) + result (p/deferred) + request-thread (p/deferred)] + (try + (let [waiter (p/thread + (p/resolve request-thread (px/current-thread)) + (p/resolve result + (try + (ffi/with-change-notifications [await-change (ffi/subscribe database)] + (p/resolve ready true) + (await-change {:timeout-ms 10000})) + (catch InterruptedException error error))))] + (try + (is (= true (p/await ready 2000 ::timeout))) + (px/interrupt (p/join request-thread)) + (is (instance? InterruptedException (p/await result 2000 ::timeout))) + (finally + (is (not= ::timeout (p/await waiter 15000 ::timeout)))))) + (ffi/with-change-notifications [await-change (ffi/subscribe database)] + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(utf8 "after-interruption") (utf8 "value")]])) + (ffi/await (ffi/commit-transaction transaction))) + (is (nil? (await-change {:timeout-ms 1000})))) + (finally + (ffi/await (ffi/close-database! database))))))) + (defn- compaction-settings [] (with-open [defaults (Settings/_default)] (-> (.toJsonString defaults) @@ -129,7 +145,7 @@ head (utf8 "head") page (utf8 "page/1")] (try - (with-open [transaction (ffi/await (ffi/begin-transaction database))] + (ffi/with-transaction [transaction database] (ffi/await (ffi/write-values transaction [[head (utf8 "r-current")] [page (utf8 "current")]])) (ffi/await (ffi/write-expiring-values transaction {:entries [[expired (utf8 "old")]] :expires-at 0})) (ffi/await (ffi/write-expiring-values transaction {:entries [[retained (utf8 "needed")]] :expires-at Long/MAX_VALUE})) @@ -137,7 +153,7 @@ (is (nil? (.expireTs (ffi/await (.getKeyValue ^Db database head))))) (is (nil? (.expireTs (ffi/await (.getKeyValue ^Db database page))))) (doseq [index (range 4)] - (with-open [transaction (ffi/await (ffi/begin-transaction database))] + (ffi/with-transaction [transaction database] (ffi/await (ffi/write-values transaction [[(utf8 (str "padding/" index)) (byte-array 4096)]])) (ffi/await (ffi/commit-transaction transaction))) (ffi/await (.flushWithOptions ^Db database (FlushOptions. FlushType/MEM_TABLE)))) @@ -150,4 +166,4 @@ (is (Arrays/equals (utf8 "needed") (ffi/await (.get ^Db database retained)))) (is (Arrays/equals (utf8 "r-current") (ffi/await (.get ^Db database head)))) (is (Arrays/equals (utf8 "current") (ffi/await (.get ^Db database page)))) - (finally (ffi/close-database! database)))))) + (finally (ffi/await (ffi/close-database! database))))))) diff --git a/test/linear/adapter/slatedb/store_test.clj b/test/linear/adapter/slatedb/store_test.clj index 7bd5754..e109ad0 100644 --- a/test/linear/adapter/slatedb/store_test.clj +++ b/test/linear/adapter/slatedb/store_test.clj @@ -10,7 +10,8 @@ [linear.adapter.slatedb.store] [linear.usecase.database.model :as model] [linear.usecase.database.revisions :as revisions] - [linear.usecase.keychain :as keychain]) + [linear.usecase.keychain :as keychain] + [promesa.core :as p]) (:import (java.util Arrays))) @@ -25,12 +26,12 @@ (revisions/initialize! store {:database (database-record keychain) :revision revision})) (defn- read-records [store record-keys] - (with-open [database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction database)] - (try - (ffi/await (ffi/read-transaction-values transaction record-keys)) - (finally - (ffi/await (ffi/rollback-transaction transaction)))))) + (connection/with-database [database store {:database-id "d-1"}] + (ffi/with-transaction [transaction database] + (try + (ffi/await (ffi/read-transaction-values transaction record-keys)) + (finally + (ffi/await (ffi/rollback-transaction transaction))))))) (deftest ^:integration initialization-encrypts-root-and-refuses-to-replace-an-existing-head (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) @@ -58,10 +59,10 @@ (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) keychain (crypto-core/new-keychain)] (try - (with-open [database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction database)] - (ffi/await (ffi/write-values transaction [[(key/head) (codec/encode-head {})]])) - (ffi/await (ffi/commit-transaction transaction))) + (connection/with-database [database store {:database-id "d-1"}] + (ffi/with-transaction [transaction database] + (ffi/await (ffi/write-values transaction [[(key/head) (codec/encode-head {})]])) + (ffi/await (ffi/commit-transaction transaction)))) (let [error (try (revisions/initialize! store {:database (database-record keychain) @@ -74,23 +75,23 @@ (is (nil? root-value))) (finally (connection/close store))))) -(deftest ^:integration failed-consistent-view-acquisition-returns-the-database-lease +(deftest ^:integration failed-consistent-view-acquisition-releases-its-use (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) keychain (crypto-core/new-keychain)] (try - (with-open [leased-database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction leased-database)] - (ffi/await - (ffi/write-values transaction [[(key/head) (byte-array [0])]])) - (ffi/await (ffi/commit-transaction transaction))) + (connection/with-database [database-handle store {:database-id "d-1"}] + (ffi/with-transaction [transaction database-handle] + (ffi/await + (ffi/write-values transaction [[(key/head) (byte-array [0])]])) + (ffi/await (ffi/commit-transaction transaction)))) (is (thrown? Exception (revisions/with-consistent-view [view store (database-record keychain)] (revisions/head view)))) - (with-open [leased-database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction leased-database)] - (is (some? transaction))) + (connection/with-database [database-handle store {:database-id "d-1"}] + (ffi/with-transaction [transaction database-handle] + (is (some? transaction)))) (finally (connection/close store))))) @@ -212,53 +213,36 @@ (finally (connection/close store))))) -(deftest ^:integration callback-failures-return-the-database-lease +(deftest ^:integration callback-failures-release-capacity (doseq [failure [(ex-info "Callback failed" {}) (AssertionError. "Callback failed")]] (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) - borrowed (atom nil) - returned (atom 0) - observed-store (assoc store - :borrow-database - (fn [id] - (let [raw ((:borrow-database store) id)] - (reset! borrowed raw) - raw)) - :return-database - (fn [id raw] - (swap! returned inc) - ((:return-database store) id raw))) database (database-record (crypto-core/new-keychain))] (try (let [caught (try - (revisions/read-consistently observed-store {:f (fn [_] (throw failure)) :database database}) + (revisions/with-consistent-view [_ store database] (throw failure)) (catch Exception error error) (catch AssertionError error error))] (is (identical? failure caught)) - (is (= 1 @returned)) - (when (= 1 @returned) - (with-open [lease (connection/database store {:database-id "d-1"})] - (is (some? lease))))) - (finally - (when (and @borrowed (zero? @returned)) - ((:return-database store) "d-1" @borrowed)) - (connection/close store)))))) + (connection/with-database [database store {:database-id "d-2"}] + (is (ffi/usable-database? database)))) + (finally (connection/close store)))))) (defn- expire-history! [store revision-ids] - (with-open [database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction database)] - (let [[value] (ffi/await (ffi/read-transaction-values transaction [(key/head)])) - head (assoc (codec/decode-head value) :history-expires-at 0)] - (doseq [revision-id revision-ids] - (ffi/await (.delete ^io.slatedb.uniffi.DbTransaction transaction (key/revision revision-id)))) - (ffi/await (ffi/write-values transaction [[(key/head) (codec/encode-head head)]])) - (ffi/await (ffi/commit-transaction transaction))))) + (connection/with-database [database store {:database-id "d-1"}] + (ffi/with-transaction [transaction database] + (let [[value] (ffi/await (ffi/read-transaction-values transaction [(key/head)])) + head (assoc (codec/decode-head value) :history-expires-at 0)] + (doseq [revision-id revision-ids] + (ffi/await (.delete ^io.slatedb.uniffi.DbTransaction transaction (key/revision revision-id)))) + (ffi/await (ffi/write-values transaction [[(key/head) (codec/encode-head head)]])) + (ffi/await (ffi/commit-transaction transaction)))))) (defn- record-expiries [store record-keys] - (with-open [database (connection/database store {:database-id "d-1"})] + (connection/with-database [database store {:database-id "d-1"}] (mapv (fn [record-key] (.expireTs ^io.slatedb.uniffi.KeyValue - (ffi/await (.getKeyValue ^io.slatedb.uniffi.Db (:raw-database database) record-key)))) - record-keys))) + (ffi/await (.getKeyValue ^io.slatedb.uniffi.Db database record-key)))) + record-keys))) (deftest ^:integration history-expires-but-current-data-survives-inactivity-and-remains-writable (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) @@ -373,3 +357,35 @@ (is (= {1 '(2)} (update-vals (:pages result) seq))) (is (thrown? clojure.lang.ExceptionInfo (model/pull database {:server-revision "r-root"}))))) (finally (connection/close store))))) + +(deftest ^:integration concurrent-publications-preserve-parent-conflicts-and-rollback + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + keychain (crypto-core/new-keychain) + database (database-record keychain) + start (p/deferred)] + (try + (seed! store keychain {:revision-id "r-root" :parent nil :database-page-count 1 :pages {1 (byte-array [0])}}) + (let [writers (mapv (fn [id] + (p/thread + (p/join start) + (try + (revisions/publish-next! store + {:database database + :revision {:revision-id id :parent "r-root" :database-page-count 1 :pages {1 (byte-array [1])}}}) + (catch Exception error error)))) + ["r-next-a" "r-next-b"])] + (try + (p/resolve start true) + (let [results (mapv #(p/await % 5000 ::timeout) writers) + successes (filter map? results) + failures (filter #(instance? Exception %) results)] + (is (= 1 (count successes))) + (is (= [::revisions/revision-conflict] (mapv #(-> % ex-data :reason) failures))) + (revisions/with-consistent-view [view store database] + (is (= (:revision-id (first successes)) (revisions/revision-id (revisions/head view)))))) + (finally + (p/resolve start true) + (doseq [writer writers] + (p/await writer 5000 ::timeout) + (p/await writer 15000 ::timeout))))) + (finally (connection/close store))))) diff --git a/test/linear/adapter/sqlite/connection_test.clj b/test/linear/adapter/sqlite/connection_test.clj index 3782eec..07120bc 100644 --- a/test/linear/adapter/sqlite/connection_test.clj +++ b/test/linear/adapter/sqlite/connection_test.clj @@ -10,70 +10,61 @@ (:import (java.util UUID))) -(defn- call-with-native-connection [f] - (let [identifier (str (UUID/randomUUID)) - path (str "/linear/" identifier ".db") - resources (vfs/install {:name (str "test-" identifier)})] - (try - (let [invocation (vfs/mount resources - {:path path - :filesystem (evaluation/evaluation (sqlite-data/snapshot - {:image (sqlite-data/sqlite-image)}) {:path path})})] - (try - (let [database (connection/open {:path path - :vfs-name (:name resources)})] - (try - (f database) - (finally - (connection/close database)))) - (finally - (vfs/unmount invocation)))) - (finally - (vfs/drain! resources) - (vfs/uninstall resources))))) +(defmacro ^:private with-connection [[binding] & body] + `(let [identifier# (str (UUID/randomUUID)) + path# (str "/linear/" identifier# ".db") + resources# (vfs/install {:name (str "test-" identifier#)})] + (try + (let [invocation# (vfs/mount resources# + {:path path# + :filesystem (evaluation/evaluation (sqlite-data/snapshot + {:image (sqlite-data/sqlite-image)}) {:path path#})})] + (try + (with-open [database# (connection/open {:path path# :vfs-name (:name resources#)})] + (let [~binding database#] ~@body)) + (finally (vfs/unmount invocation#)))) + (finally + (vfs/drain! resources#) + (vfs/uninstall resources#))))) (deftest ^:integration execute-translates-native-sqlite-errors - (call-with-native-connection - (fn [database] - (let [failure (try - (connection/execute database "not valid SQL") - nil - (catch clojure.lang.ExceptionInfo error - error)) - data (ex-data failure)] - (is (= ::anomaly/fault (::anomaly/category data))) - (is (= ::connection/statement-failed (:reason data))) - (is (pos-int? (:code data))) - (is (pos-int? (:extended-code data))) - (is (string/includes? (or (::anomaly/message data) "") "syntax error")))))) + (with-connection [database] + (let [failure (try + (connection/execute database "not valid SQL") + nil + (catch clojure.lang.ExceptionInfo error + error)) + data (ex-data failure)] + (is (= ::anomaly/fault (::anomaly/category data))) + (is (= ::connection/statement-failed (:reason data))) + (is (pos-int? (:code data))) + (is (pos-int? (:extended-code data))) + (is (string/includes? (or (::anomaly/message data) "") "syntax error"))))) (deftest ^:integration execute-rejects-a-closed-connection - (call-with-native-connection - (fn [database] - (connection/close database) - (let [failure (try - (connection/execute database "SELECT 1") - nil - (catch clojure.lang.ExceptionInfo error - error))] - (is (= ::connection/connection-closed - (:reason (ex-data failure)))))))) + (with-connection [database] + (connection/close database) + (let [failure (try + (connection/execute database "SELECT 1") + nil + (catch clojure.lang.ExceptionInfo error + error))] + (is (= ::connection/connection-closed + (:reason (ex-data failure))))))) (deftest ^:integration constraint-error-survives-statement-finalization - (call-with-native-connection - (fn [database] - (connection/execute database "PRAGMA locking_mode = EXCLUSIVE") - (connection/execute database "CREATE TABLE unique_ids (id INTEGER PRIMARY KEY); INSERT INTO unique_ids VALUES (1)") - (let [failure (try - (connection/execute-statement database {:sql "INSERT INTO unique_ids VALUES (1)" :parameters []}) - nil - (catch clojure.lang.ExceptionInfo error error))] - (is (= ::connection/statement-failed (:reason (ex-data failure)))) - (is (string/includes? (or (::anomaly/message (ex-data failure)) "") "UNIQUE constraint failed")) - (is (= [1] (connection/query-integers database {:sql "SELECT COUNT(*) FROM unique_ids" :parameters []}))))))) + (with-connection [database] + (connection/execute database "PRAGMA locking_mode = EXCLUSIVE") + (connection/execute database "CREATE TABLE unique_ids (id INTEGER PRIMARY KEY); INSERT INTO unique_ids VALUES (1)") + (let [failure (try + (connection/execute-statement database {:sql "INSERT INTO unique_ids VALUES (1)" :parameters []}) + nil + (catch clojure.lang.ExceptionInfo error error))] + (is (= ::connection/statement-failed (:reason (ex-data failure)))) + (is (string/includes? (or (::anomaly/message (ex-data failure)) "") "UNIQUE constraint failed")) + (is (= [1] (connection/query-integers database {:sql "SELECT COUNT(*) FROM unique_ids" :parameters []})))))) (deftest ^:integration close-is-idempotent - (call-with-native-connection - (fn [database] - (connection/close database) - (is (nil? (connection/close database)))))) + (with-connection [database] + (connection/close database) + (is (nil? (connection/close database))))) diff --git a/test/linear/adapter/sqlite/vfs_test.clj b/test/linear/adapter/sqlite/vfs_test.clj index 54300cb..50f91f3 100644 --- a/test/linear/adapter/sqlite/vfs_test.clj +++ b/test/linear/adapter/sqlite/vfs_test.clj @@ -99,9 +99,9 @@ (let [resources (resources) invocation (vfs/mount resources {:path "/test.db" :filesystem (filesystem)})] (vfs/drain! resources) - (is (= ::waiting (p/await (:drained @(:state resources)) 10 ::waiting))) + (is (not (p/done? (:drained @(:state resources))))) (vfs/unmount invocation) - (is (= true (p/await (:drained @(:state resources)) 1000 ::timeout))))) + (is (p/done? (:drained @(:state resources)))))) (deftest draining-rejects-new-invocations (let [resources (resources)] diff --git a/test/linear/e2e.clj b/test/linear/e2e.clj index 4b8a312..19d9dd0 100644 --- a/test/linear/e2e.clj +++ b/test/linear/e2e.clj @@ -87,7 +87,8 @@ system (test/run {:config (config) :keys #{:duct.server.http/jetty} :profiles [:test :main] - :vars (assoc (jwt/oidc-vars fixture) 'port 0)})] + :vars (assoc (jwt/oidc-vars fixture) + 'port 0 'max-long-poll-timeout-ms 5000)})] (try (run-client! (server-url system) {"LINEAR_E2E_MANAGER_TOKEN" (jwt/access-token fixture {:subject manager}) diff --git a/test/linear/handler/turso/protobuf_test.clj b/test/linear/handler/turso/protobuf_test.clj index 00fb3dc..a85a2c2 100644 --- a/test/linear/handler/turso/protobuf_test.clj +++ b/test/linear/handler/turso/protobuf_test.clj @@ -81,6 +81,16 @@ 0x12 0x02 0x01 0x02)) (seq (.toByteArray output)))))) +(deftest long-poll-timeout-is-an-unsigned-32-bit-value + (is (= 0xffffffff + (:long-poll-timeout-ms + (protobuf/decode-pull (byte-array (map unchecked-byte [0x20 0xff 0xff 0xff 0xff 0x0f])))))) + (doseq [value [[0x20 0x80 0x80 0x80 0x80 0x10] + [0x20 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0xff 0x01]]] + (is (= ::protobuf/malformed-protobuf + (:reason (test/catch-ex-data + #(protobuf/decode-pull (byte-array (map unchecked-byte value))))))))) + (deftest protobuf-parser-rejects-truncated-length-delimited-fields (testing "syntax errors are codec errors, not semantic validation" (is (thrown? Exception diff --git a/test/linear/handler/turso/pull_test.clj b/test/linear/handler/turso/pull_test.clj index 9a91633..7b107af 100644 --- a/test/linear/handler/turso/pull_test.clj +++ b/test/linear/handler/turso/pull_test.clj @@ -21,7 +21,6 @@ (deftest unsupported-pull-options-are-rejected-before-database-resolution (doseq [[option bytes] [[:encoding [0x08 0x01]] [:stream-kind [0x40 0x01]] - [:long-poll-timeout-ms [0x20 0x01]] [:server-query-selector [0x3a 0x01 0x71]] [:client-pages [0x32 0x01 0x01]]]] (is (= 400 (:status ((pull/handler {}) {:body (byte-array bytes)}))) diff --git a/test/linear/usecase/database/pull_test.clj b/test/linear/usecase/database/pull_test.clj index a9b8dab..f6330e3 100644 --- a/test/linear/usecase/database/pull_test.clj +++ b/test/linear/usecase/database/pull_test.clj @@ -3,6 +3,7 @@ [clojure.test :refer [deftest is]] [duct.test :refer [with-system]] [linear.adapter.postgres] + [linear.adapter.slatedb.connection :as connection] [linear.adapter.slatedb.store] [linear.test :refer [run]] [linear.test-data.postgres :as postgres-data] @@ -11,7 +12,9 @@ [linear.usecase.database :as database] [linear.usecase.database.revisions :as revisions] [linear.usecase.transaction :as transaction] - [linear.usecase.vault :as vault])) + [linear.usecase.vault :as vault] + [promesa.core :as p] + [promesa.exec :as px])) (deftest ^:integration pull-traverses-postgres-vault-decryption-and-slatedb (with-system [system (run {:keys [:duct.database/sql @@ -99,3 +102,115 @@ (is (thrown? clojure.lang.ExceptionInfo (database/pull ctx pull-args))) (restore-fn ctx (assoc args :token token)) (is (= (seq page) (seq (get-in (database/pull ctx pull-args) [:pages 1]))))))) + +(deftest ^:integration long-poll-caps-waiting-and-closes-read-scopes-before-awaiting + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime + :linear.adapter.slatedb.store/store]})] + (let [datasource (:duct.database.sql/hikaricp system) + store (:linear.adapter.slatedb.store/store system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + transacting? (atom false) + viewing? (atom false) + waits (atom []) + transactable (reify transaction/Transactable + (-transact [_ f options] + (reset! transacting? true) + (try + (transaction/-transact datasource f options) + (finally (reset! transacting? false))))) + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] + (revisions/-read-consistently store + (fn [view] + (reset! viewing? true) + (try (f view) (finally (reset! viewing? false)))) + database)) + revisions/RevisionAwaitable + (-await-revision [_ args] + (is (false? @viewing?)) + (is (false? @transacting?)) + (swap! waits conj args) + nil)) + context {::core/database transactable ::core/key-service key-protection + ::core/revision-store reader ::core/max-long-poll-timeout-ms 40} + args {:actor "actor-1" :database-id database-id}] + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (doseq [options [{:long-poll-timeout-ms 5000} + {:client-revision "r-root" :long-poll-timeout-ms 0} + {:client-revision "r-root" :server-revision "r-root" :long-poll-timeout-ms 5000}]] + (is (= "r-root" (:server-revision (database/pull context (merge args options)))))) + (is (empty? @waits)) + (doseq [timeout [10 5000 0xffffffff]] + (is (= {} (:pages (database/pull context (assoc args :client-revision "r-root" + :long-poll-timeout-ms timeout)))))) + (is (= [{:database-id database-id :revision-id "r-root" :timeout-ms 10} + {:database-id database-id :revision-id "r-root" :timeout-ms 40} + {:database-id database-id :revision-id "r-root" :timeout-ms 40}] @waits)) + (revisions/publish-next! store + {:database {:id database-id :vault {:keychain keychain}} + :revision {:revision-id "r-next" :parent "r-root" :database-page-count 1 :pages {1 (byte-array [7])}}}) + (let [result (database/pull context (assoc args :client-revision "r-root" :long-poll-timeout-ms 5000))] + (is (= "r-next" (:server-revision result))) + (is (= {1 '(7)} (update-vals (:pages result) seq))) + (is (= 3 (count @waits))))))) + +(deftest ^:integration long-poll-rechecks-database-access-after-waiting + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime + :linear.adapter.slatedb.store/store]})] + (let [datasource (:duct.database.sql/hikaricp system) + store (:linear.adapter.slatedb.store/store system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + context {::core/database datasource ::core/key-service key-protection + ::core/revision-store store ::core/max-long-poll-timeout-ms 5000} + args {:actor "actor-1" :database-id database-id} + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] (revisions/-read-consistently store f database)) + revisions/RevisionAwaitable + (-await-revision [_ _] + (database/close! context args) + nil))] + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (is (= ::database/database-not-found + (:reason (linear.test/catch-ex-data + #(database/pull (assoc context ::core/revision-store reader) + (assoc args :client-revision "r-root" :long-poll-timeout-ms 100))))))))) + +(deftest ^:integration eviction-wakes-long-poll-and-subsequent-pull-still-works + (with-system [system (run {:keys [:duct.database/sql :duct.migrator/ragtime]})] + (let [datasource (:duct.database.sql/hikaricp system) + {:keys [database-id key-protection keychain]} + (postgres-data/create-database! {:datasource datasource}) + store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + entering-wait (p/deferred) + reader (reify + revisions/ConsistentReadable + (-read-consistently [_ f database] + (revisions/-read-consistently store f database)) + revisions/RevisionAwaitable + (-await-revision [_ args] + (p/resolve entering-wait true) + (revisions/-await-revision store args))) + context {::core/database datasource ::core/key-service key-protection + ::core/revision-store reader ::core/max-long-poll-timeout-ms 10000} + args {:actor "actor-1" :database-id database-id}] + (try + (slatedb-data/store-root! {:store store :database-id database-id :keychain keychain :pages {}}) + (let [poll (p/thread (database/pull context (assoc args :client-revision "r-root" :long-poll-timeout-ms 10000)))] + (try + (is (= true (p/await entering-wait 5000 ::timeout))) + ;; Repeat pressure until the real adapter has installed its subscription. + (loop [attempts 200] + (when (and (pos? attempts) (not (p/done? poll))) + (connection/with-database [_ store {:database-id "other"}]) + (px/sleep 10) + (recur (dec attempts)))) + (let [result (p/await poll 1000 ::timeout)] + (is (= "r-root" (:server-revision result))) + (is (= {} (:pages result)))) + (finally (p/await poll 15000 ::timeout)))) + (is (= "r-root" (:server-revision (database/pull context args)))) + (finally (connection/close store))))))