diff --git a/compose.yaml b/compose.yaml index 630a7c4..d512cde 100644 --- a/compose.yaml +++ b/compose.yaml @@ -26,6 +26,7 @@ services: environment: JDBC_DATABASE_URL: jdbc:postgresql://postgres:5432/linear?user=postgres&password=postgres SLATEDB_OBJECT_STORE_URL: memory:/// + REVISION_RETENTION_DAYS: LINEAR_DEPLOYMENT_TARGET: GCP_KMS_KEY_NAME: OIDC_ISSUER: diff --git a/docs/architecture.md b/docs/architecture.md index 7e24a26..d7774c4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -177,6 +177,21 @@ column. SlateDB owns revisions, pages, and HEAD and publishes them atomically in its own transaction, rejecting stale parents. Push reads PostgreSQL and writes SlateDB; it does not introduce dual writes or a cross-store transaction. +Evaluators publish changed page after-images. The SlateDB adapter captures +before-images in the publication transaction and stores encrypted undo records +with the parent HEAD descriptor. HEAD contains its own page count, and neither +HEAD nor current pages expire. Historical reads apply undo records from HEAD +back to the requested revision, stopping before undoing that revision. +Physical tail pages remain available across database shrinkage and regrowth; +snapshots hide pages beyond their logical page count. + +Undo records expire after the configured history retention (30 days by +default), measured from the update that replaced the previous state. Expiry +times never decrease along the revision chain. SlateDB owns compaction and +physical garbage collection. An expired client cursor triggers a full-page +pull for the selected target; an unavailable explicit target or missing +unexpired history remains an error. + ## PostgreSQL loading `linear.adapter.postgres` is a load-only technology root. It loads diff --git a/docs/deployment.md b/docs/deployment.md index 22cab5b..86ac7d9 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -22,6 +22,13 @@ tokens must use RS256 and include the configured issuer, audience, expiration, and a nonempty `sub`. See [authentication](authentication.md). `PORT` optionally changes the listening port. +`REVISION_RETENTION_DAYS` sets revision history retention in days (default: 30). +HEAD and current pages never expire, including for inactive databases. Undo +records retain the previous state for at least this period after an update; +SlateDB removes expired records during compaction. Retention changes apply to +future writes and cannot recover history already removed. A shorter setting +does not let a newer undo record expire before its predecessor. + Use a dedicated GCS bucket. A URL path such as `gs://bucket/prefix` does not provide application-level prefix isolation; use the bucket-only URL. Database IDs supply the object paths. Keep the same KMS key and its old decryptable key diff --git a/duct.edn b/duct.edn index be0cd8c..7aca0da 100644 --- a/duct.edn +++ b/duct.edn @@ -8,6 +8,9 @@ slatedb-max-open-databases {:env "SLATEDB_MAX_OPEN_DATABASES" :type :int :default 128} + revision-retention-days {:env "REVISION_RETENTION_DAYS" + :type :int + :default 30} sqlite-vfs-name {:env "SQLITE_VFS_NAME" :default "linear-sqlite"} deployment-target {:env "LINEAR_DEPLOYMENT_TARGET" @@ -24,7 +27,8 @@ :linear.adapter.slatedb.store/store {:object-store-url #ig/var slatedb-object-store-url - :max-open-databases #ig/var slatedb-max-open-databases} + :max-open-databases #ig/var slatedb-max-open-databases + :revision-retention-days #ig/var revision-retention-days} :linear.adapter.postgres/module {} diff --git a/src/linear/adapter/slatedb/connection.clj b/src/linear/adapter/slatedb/connection.clj index f6eb4e2..41a6f71 100644 --- a/src/linear/adapter/slatedb/connection.clj +++ b/src/linear/adapter/slatedb/connection.clj @@ -7,7 +7,7 @@ (org.apache.commons.pool2.impl DefaultPooledObject GenericKeyedObjectPool GenericKeyedObjectPoolConfig))) (defrecord Connection - [borrow-database return-database close-resources]) + [borrow-database return-database close-resources revision-retention-days]) (defn connection [options] (map->Connection options)) @@ -60,14 +60,17 @@ {:malli/schema [:-> [:map [:object-store-url :string] - [:max-open-databases pos-int?]] + [:max-open-databases pos-int?] + [:revision-retention-days {:optional true} pos-int?]] [:fn #(instance? Connection %)]]} - [{:keys [object-store-url max-open-databases]}] + [{: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 - {:borrow-database (fn [database-id] + {: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 diff --git a/src/linear/adapter/slatedb/ffi.clj b/src/linear/adapter/slatedb/ffi.clj index 33b806f..e7974f1 100644 --- a/src/linear/adapter/slatedb/ffi.clj +++ b/src/linear/adapter/slatedb/ffi.clj @@ -5,7 +5,7 @@ [promesa.core :as p] [promesa.exec :as px]) (:import - (io.slatedb.uniffi CloseReason Db DbBuilder DbSnapshot DbTransaction Error$Closed Error$Data Error$Internal Error$Invalid Error$Transaction Error$Unavailable IsolationLevel ObjectStore) + (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) (java.util.concurrent CompletionException ExecutionException))) (defn- closed-category [^Error$Closed error] @@ -68,12 +68,12 @@ (p/resolved []) keys)) -(defn- write-values-impl [transaction entries] +(defn- write-values-impl [transaction entries options] (reduce (fn [write-promise [key value]] (p/then write-promise (fn [_previous-write] - (invoke :write-values #(.put ^DbTransaction transaction key value))))) + (invoke :write-values #(.putWithOptions ^DbTransaction transaction key value options))))) (p/resolved nil) entries)) @@ -147,10 +147,17 @@ keys)) (defn write-values [transaction entries] - (write-values-impl transaction entries)) + (write-values-impl transaction entries (PutOptions. (Ttl$NoExpiry.)))) + +(defn write-expiring-values [transaction {:keys [entries expires-at]}] + (write-values-impl transaction entries (PutOptions. (Ttl$ExpireAtMillis. (long expires-at))))) (defn commit-transaction [transaction] - (invoke :commit-transaction #(.commit ^DbTransaction transaction))) + (p/then (invoke :commit-transaction #(.commit ^DbTransaction transaction)) + (fn [handle] + (-> (invoke :commit-transaction #(.awaitDurable ^WriteHandle handle)) + (p/finally (fn [_value _error] + (.close ^WriteHandle handle))))))) (defn rollback-transaction [transaction] (invoke :rollback-transaction #(.rollback ^DbTransaction transaction))) diff --git a/src/linear/adapter/slatedb/snapshot.clj b/src/linear/adapter/slatedb/snapshot.clj index 5c96ab3..61eef46 100644 --- a/src/linear/adapter/slatedb/snapshot.clj +++ b/src/linear/adapter/slatedb/snapshot.clj @@ -3,142 +3,122 @@ [cognitect.anomalies :as anomaly] [linear.adapter.slatedb.codec :as codec] [linear.adapter.slatedb.key :as key] - [linear.usecase.database.revisions :as revisions] - [linear.usecase.keychain :as keychain])) + [linear.usecase.database.revisions :as revisions])) -(defn head-revision-id - {:malli/schema [:-> [:fn ifn?] ::revisions/revision-id]} - [read-values] +(defn head-metadata [read-values] (let [[value] (read-values [(key/head)]) head (codec/decode-head value)] - (or (:revision-id head) - (throw (ex-info "SlateDB HEAD does not exist" - {::anomaly/category ::anomaly/not-found - :reason ::revisions/head-not-found}))))) + (if (:revision-id head) + head + (throw (ex-info "SlateDB HEAD does not exist" + {::anomaly/category ::anomaly/not-found + :reason ::revisions/head-not-found}))))) -(defn- revision-metadata [read-values keychain revision-id] - (let [record-key (key/revision revision-id) +(defn- read-undo-record [read-values keychain head] + (let [revision-id (:revision-id head) + record-key (key/revision revision-id) [value] (read-values [record-key])] - (codec/decode-revision keychain {:record-key record-key :value value}))) + (or (codec/decode-revision keychain {:record-key record-key :value value}) + (let [expired? (<= (:history-expires-at head) (System/currentTimeMillis))] + (throw (ex-info "SlateDB revision does not exist" + {::anomaly/category (if expired? ::anomaly/not-found ::anomaly/fault) + :reason (if expired? ::revisions/revision-not-found ::revisions/incomplete-revision) + :expired? expired? + :revision-id revision-id})))))) + +(defn- ensure-unvisited! [seen revision-id] + (when (contains? seen revision-id) + (throw (ex-info "Revision chain contains a cycle" + {::anomaly/category ::anomaly/fault + :reason ::revisions/revision-cycle + :revision-id revision-id})))) -(defn- pages-at-revision [read-values keychain revision-id page-ids] - (loop [revision-id revision-id - page-ids (set page-ids) - pages {} - seen #{}] +(defn- state-at-revision [read-values keychain head target-revision-id restore-pages?] + (loop [head head + pages {} + seen #{}] (cond - (empty? page-ids) pages - (nil? revision-id) - (throw (ex-info "Revision does not contain page state" - {::anomaly/category ::anomaly/fault - :reason ::revisions/incomplete-revision - :page-ids page-ids})) - (contains? seen revision-id) - (throw (ex-info "Revision chain contains a cycle" - {::anomaly/category ::anomaly/fault - :reason ::revisions/revision-cycle - :revision-id revision-id})) + (nil? head) + (throw (ex-info "SlateDB revision does not exist" + {::anomaly/category ::anomaly/not-found + :reason ::revisions/revision-not-found + :revision-id target-revision-id})) + + (= target-revision-id (:revision-id head)) + {:metadata head :undo-pages pages} + :else - (if-let [revision (revision-metadata read-values keychain revision-id)] - (let [found (select-keys (:pages revision) page-ids)] + (do + (ensure-unvisited! seen (:revision-id head)) + (let [revision (read-undo-record read-values keychain head)] (recur (:parent revision) - (reduce disj page-ids (keys found)) - (merge pages found) - (conj seen revision-id))) - (throw (ex-info "SlateDB revision does not exist" - {::anomaly/category ::anomaly/not-found - :reason ::revisions/revision-not-found - :revision-id revision-id})))))) + (if restore-pages? (merge pages (:undo-pages revision)) pages) + (conj seen (:revision-id head)))))))) + +(defn- changed-page-ids [read-values keychain head client-revision-id] + (loop [head head + page-ids #{} + seen #{}] + (cond + (nil? head) + (throw (ex-info "Client revision is not an ancestor of the target revision" + {::anomaly/category ::anomaly/incorrect + :reason ::revisions/invalid-revision-cursor + :client-revision-id client-revision-id})) + + (= client-revision-id (:revision-id head)) page-ids -(defn- changed-page-ids [read-values keychain target-revision-id client-revision-id] - (loop [revision-id target-revision-id - page-ids #{} - seen #{}] - (if (= client-revision-id revision-id) - page-ids - (cond - (nil? revision-id) - (throw (ex-info "Client revision is not an ancestor of the target revision" - {::anomaly/category ::anomaly/incorrect - :reason ::revisions/invalid-revision-cursor - :client-revision-id client-revision-id - :target-revision-id target-revision-id})) - (contains? seen revision-id) - (throw (ex-info "Revision chain contains a cycle" - {::anomaly/category ::anomaly/fault - :reason ::revisions/revision-cycle - :revision-id revision-id})) - :else - (if-let [revision (revision-metadata read-values keychain revision-id)] + :else + (do + (ensure-unvisited! seen (:revision-id head)) + (let [revision (read-undo-record read-values keychain head)] (recur (:parent revision) - (into page-ids (keys (:pages revision))) - (conj seen revision-id)) - (throw (ex-info "Client revision is not an ancestor of the target revision" - {::anomaly/category ::anomaly/incorrect - :reason ::revisions/invalid-revision-cursor - :client-revision-id client-revision-id - :target-revision-id target-revision-id}))))))) + (into page-ids (keys (:undo-pages revision))) + (conj seen (:revision-id head)))))))) -(defn- current-pages [read-values keychain page-ids] - (let [page-ids (vec page-ids) - record-keys (mapv key/page page-ids) - values (read-values record-keys)] +(defn- fetch-pages [read-values keychain {:keys [metadata undo-pages]} page-ids] + (let [page-count (:database-page-count metadata) + current-ids (vec (filter #(and (<= % page-count) + (not (contains? undo-pages %))) + page-ids)) + record-keys (mapv key/page current-ids) + values (read-values record-keys) + current-pages (into {} + (map (fn [page-id record-key value] + [page-id (codec/decode-page keychain {:record-key record-key :value value})]) + current-ids record-keys values)) + pages (merge current-pages (select-keys undo-pages page-ids))] (into {} - (map (fn [page-id record-key value] - [page-id (codec/decode-page keychain {:record-key record-key :value value})]) - page-ids - record-keys - values)))) + (map (fn [page-id] + (let [page (when (<= page-id page-count) (get pages page-id))] + (when (and (<= page-id page-count) (nil? page)) + (throw (ex-info "Revision does not contain page state" + {::anomaly/category ::anomaly/fault + :reason ::revisions/incomplete-revision + :revision-id (:revision-id metadata) + :page-id page-id}))) + [page-id page]))) + page-ids))) -(defrecord ^:private Snapshot [read-values revision-id keychain fetch-pages] +(defrecord ^:private Snapshot [read-values keychain state] revisions/Snapshot (-revision-id [_] - revision-id) + (get-in state [:metadata :revision-id])) (-size [_] - (:database-page-count - (revision-metadata read-values keychain revision-id))) + (get-in state [:metadata :database-page-count])) (-fetch-pages-by-ids [_ {:keys [ids]}] - (fetch-pages ids))) - -(defn- create-snapshot [read-values revision-id keychain fetch-pages] - (->Snapshot read-values revision-id keychain fetch-pages)) - -(defn snapshot - {:malli/schema [:-> - [:map - [:read-values [:fn ifn?]] - [:revision-id ::revisions/revision-id] - [:keychain [:fn keychain/keychain?]]] - ::revisions/snapshot]} - [{:keys [read-values revision-id keychain]}] - (create-snapshot read-values - revision-id - keychain - #(current-pages read-values keychain %))) + (fetch-pages read-values keychain state ids))) (defrecord ^:private ConsistentView [read-values keychain] revisions/ConsistentView (-head [_] - (let [revision-id (head-revision-id read-values)] - (create-snapshot read-values - revision-id - keychain - #(current-pages read-values keychain %)))) + (->Snapshot read-values keychain {:metadata (head-metadata read-values) :undo-pages {}})) (-as-of [_ revision-id] - (when-not (revision-metadata read-values keychain revision-id) - (throw (ex-info "SlateDB revision does not exist" - {::anomaly/category ::anomaly/not-found - :reason ::revisions/revision-not-found - :revision-id revision-id}))) - (create-snapshot read-values - revision-id - keychain - #(pages-at-revision read-values keychain revision-id %))) + (->Snapshot read-values keychain + (state-at-revision read-values keychain (head-metadata read-values) revision-id true))) (-changes-since [_ target-revision-id client-revision-id] - (changed-page-ids read-values - keychain - target-revision-id - client-revision-id))) + (let [state (state-at-revision read-values keychain (head-metadata read-values) target-revision-id false)] + (changed-page-ids read-values keychain (:metadata state) client-revision-id)))) (defn consistent-view [{:keys [read-values keychain]}] (->ConsistentView read-values keychain)) diff --git a/src/linear/adapter/slatedb/store.clj b/src/linear/adapter/slatedb/store.clj index fe64782..9da731f 100644 --- a/src/linear/adapter/slatedb/store.clj +++ b/src/linear/adapter/slatedb/store.clj @@ -12,17 +12,27 @@ (io.slatedb.uniffi DbTransaction) (java.lang AutoCloseable))) -(defn- revision-records [keychain revision] - (let [revision-id (:revision-id revision) - revision-key (key/revision revision-id)] - (into [[revision-key - (codec/encode-revision keychain {:record-key revision-key :revision revision})]] - (concat - (map (fn [[page-id page]] - (let [page-key (key/page page-id)] - [page-key (codec/encode-page keychain {:record-key page-key :page page})])) - (:pages revision)) - [[(key/head) (codec/encode-head {:revision-id revision-id})]])))) +(defn- current-state-records [keychain revision head] + (conj (mapv (fn [[page-id page]] + (let [page-key (key/page page-id)] + [page-key (codec/encode-page keychain {:record-key page-key :page page})])) + (:pages revision)) + [(key/head) (codec/encode-head head)])) + +(defn- undo-pages [read-values keychain head page-ids] + (let [record-keys (mapv key/page page-ids) + values (read-values record-keys)] + (into {} + (map (fn [page-id record-key value] + (let [page (codec/decode-page keychain {:record-key record-key :value value})] + (when (and head (<= page-id (:database-page-count head)) (nil? page)) + (throw (ex-info "Revision does not contain page state" + {::anomaly/category ::anomaly/fault + :reason ::revisions/incomplete-revision + :page-id page-id}))) + ;; Keep physical tail pages: an older revision can still need them after a regrowth. + [page-id page])) + page-ids record-keys values)))) (defn- add-suppressed! [error suppressed] (when suppressed @@ -65,11 +75,23 @@ (when value (throw (revision-conflict "Database HEAD already exists" {} nil))) nil) - (snapshot/head-revision-id read-values))] - (ensure-current-parent! head revision) + (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-values transaction - (revision-records keychain revision))) + (current-state-records keychain revision next-head))) (ffi/await (ffi/commit-transaction transaction)) revision) (catch Exception error diff --git a/src/linear/usecase/database/model.clj b/src/linear/usecase/database/model.clj index 28f1307..d12483a 100644 --- a/src/linear/usecase/database/model.clj +++ b/src/linear/usecase/database/model.clj @@ -77,6 +77,19 @@ :actual-parent (:parent revision)}))) revision)))) +(defn- changed-page-ids [view target-revision client-revision page-count] + (cond + (not client-revision) (set (range 1 (inc page-count))) + (= client-revision target-revision) #{} + :else + (try + (revisions/changes-since view {:target-revision-id target-revision :client-revision-id client-revision}) + (catch clojure.lang.ExceptionInfo error + (let [data (ex-data error)] + (if (and (= ::revisions/revision-not-found (:reason data)) (:expired? data)) + (set (range 1 (inc page-count))) + (throw error))))))) + (defn pull {:malli/schema [:-> [:and ::database @@ -96,13 +109,11 @@ (revisions/head view)) target-revision (revisions/revision-id snapshot) page-count (revisions/size snapshot) - changed-page-ids (cond - (not client-revision) (set (range 1 (inc page-count))) - (= client-revision target-revision) #{} - :else (revisions/changes-since view {:target-revision-id target-revision :client-revision-id client-revision})) - selected-page-ids (if page-ids - (set (filter page-ids changed-page-ids)) - changed-page-ids)] + changed-page-ids (changed-page-ids view target-revision client-revision page-count) + selected-page-ids (into #{} + (filter #(and (<= % page-count) + (or (nil? page-ids) (contains? page-ids %)))) + changed-page-ids)] {:server-revision target-revision :database-page-count page-count :pages (revisions/fetch-pages-by-ids snapshot {:ids selected-page-ids})})) diff --git a/test/linear/adapter/slatedb/codec_test.clj b/test/linear/adapter/slatedb/codec_test.clj index 44228f9..67fe372 100644 --- a/test/linear/adapter/slatedb/codec_test.clj +++ b/test/linear/adapter/slatedb/codec_test.clj @@ -44,16 +44,15 @@ (:reason (test/catch-ex-data #(codec/decode-page keychain {:record-key target :value ciphertext}))))))) -(deftest revision-codec-retains-changed-pages +(deftest revision-codec-retains-before-images-and-absent-pages (let [keychain (fake-keychain :primary) record-key (key/revision "r-01K002") - revision {:revision-id "r-01K002" - :parent "r-01K001" - :database-page-count 1 - :pages {1 (byte-array [1])}} + revision {:parent {:revision-id "r-01K001" :database-page-count 1 :history-expires-at 123} + :undo-pages {1 (byte-array [1]) 2 nil}} decoded (codec/decode-revision keychain {:record-key record-key :value (codec/encode-revision keychain {:record-key record-key :revision revision})})] - (is (= (dissoc revision :pages) (dissoc decoded :pages))) - (is (= #{1} (set (keys (:pages decoded))))) - (is (Arrays/equals (get-in revision [:pages 1]) - (get-in decoded [:pages 1]))))) + (is (= (:parent revision) (:parent decoded))) + (is (= #{1 2} (set (keys (:undo-pages decoded))))) + (is (nil? (get-in decoded [:undo-pages 2]))) + (is (Arrays/equals (get-in revision [:undo-pages 1]) + (get-in decoded [:undo-pages 1]))))) diff --git a/test/linear/adapter/slatedb/ffi_test.clj b/test/linear/adapter/slatedb/ffi_test.clj index f15dfc8..cd53944 100644 --- a/test/linear/adapter/slatedb/ffi_test.clj +++ b/test/linear/adapter/slatedb/ffi_test.clj @@ -1,11 +1,12 @@ (ns linear.adapter.slatedb.ffi-test (:require + [cheshire.core :as json] [clojure.test :refer [deftest is testing]] [cognitect.anomalies :as anomaly] [linear.adapter.slatedb.ffi :as ffi] [promesa.core :as p]) (:import - (io.slatedb.uniffi CloseReason Error$Closed Error$Data Error$Internal Error$Invalid Error$Transaction Error$Unavailable) + (io.slatedb.uniffi CloseReason Db DbBuilder Error$Closed Error$Data Error$Internal Error$Invalid Error$Transaction Error$Unavailable FlushOptions FlushType Settings) (java.nio.charset StandardCharsets) (java.util Arrays))) @@ -103,3 +104,49 @@ (finally (ffi/close-database! database) (ffi/close-object-store! object-store))))) + +(defn- compaction-settings [] + (with-open [defaults (Settings/_default)] + (-> (.toJsonString defaults) + (json/parse-string true) + (assoc :default_ttl_millis 1) + (assoc-in [:compactor_options :poll_interval] "10ms") + (assoc-in [:compactor_options :commit_compacted_interval] "10ms") + (assoc-in [:compactor_options :worker :compactions_poll_interval] "10ms") + (assoc-in [:compactor_options :scheduler_options] {"min_compaction_sources" "2" "max_compaction_sources" "2"}) + json/generate-string + Settings/fromJsonString))) + +(deftest ^:integration compaction-removes-expired-history-and-preserves-current-and-retained-values + (with-open [object-store (ffi/open-object-store "memory:///") + settings (compaction-settings) + builder (DbBuilder. "d-ttl-compaction" object-store)] + (.withSettings builder settings) + (let [database (ffi/await (.build builder)) + expired (utf8 "revision/expired") + retained (utf8 "revision/retained") + head (utf8 "head") + page (utf8 "page/1")] + (try + (with-open [transaction (ffi/await (ffi/begin-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})) + (ffi/await (ffi/commit-transaction transaction))) + (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/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)))) + (let [deadline (+ (System/nanoTime) 10000000000)] + (loop [] + (when (and (some? (ffi/await (.get ^Db database expired))) (< (System/nanoTime) deadline)) + (Thread/sleep 20) + (recur)))) + (is (nil? (ffi/await (.get ^Db database expired)))) + (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)))))) diff --git a/test/linear/adapter/slatedb/snapshot_test.clj b/test/linear/adapter/slatedb/snapshot_test.clj index aefe1ab..653ac68 100644 --- a/test/linear/adapter/slatedb/snapshot_test.clj +++ b/test/linear/adapter/slatedb/snapshot_test.clj @@ -1,179 +1,101 @@ (ns linear.adapter.slatedb.snapshot-test (:require [clojure.test :refer [deftest is]] + [cognitect.anomalies :as anomaly] [linear.adapter.crypto.core :as crypto-core] [linear.adapter.slatedb.codec :as codec] - [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 - (java.util Arrays UUID))) + [linear.test :refer [catch-ex-data]] + [linear.usecase.database.revisions :as revisions])) -(defn- revision-records [keychain] - (let [revision {:revision-id "r-01K002" - :parent "r-01K001" - :database-page-count 2 - :pages {1 (byte-array [2]) - 2 (byte-array [9])}} - revision-key (key/revision (:revision-id revision))] - [[(key/head) (codec/encode-head {:revision-id (:revision-id revision)})] - [revision-key (codec/encode-revision keychain {:record-key revision-key :revision revision})] - [(key/page 1) (codec/encode-page keychain {:record-key (key/page 1) :page (get-in revision [:pages 1])})] - [(key/page 2) (codec/encode-page keychain {:record-key (key/page 2) :page (get-in revision [:pages 2])})]])) +(defn- head [revision-id size] + {:revision-id revision-id :database-page-count size :history-expires-at Long/MAX_VALUE}) -(defn- with-seeded-snapshot [f] +(defn- view [{:keys [head history pages]}] (let [keychain (crypto-core/new-keychain) - object-store (ffi/open-object-store "memory:///") - database (ffi/open-database! object-store (str "d-" (UUID/randomUUID)))] - (try - (let [transaction (ffi/await (ffi/begin-transaction database))] - (try - (ffi/await (ffi/write-values transaction (revision-records keychain))) - (ffi/await (ffi/commit-transaction transaction)) - (finally - (ffi/close-transaction! transaction)))) - (let [snapshot (ffi/await (ffi/open-snapshot database))] - (try - (f snapshot keychain) - (finally - (ffi/close-snapshot! snapshot)))) - (finally - (ffi/close-database! database) - (ffi/close-object-store! object-store))))) + records (into {(seq (key/head)) (codec/encode-head head)} + (concat + (map (fn [[revision-id undo]] + (let [record-key (key/revision revision-id)] + [(seq record-key) (codec/encode-revision keychain {:record-key record-key :revision undo})])) + history) + (map (fn [[page-id page]] + (let [record-key (key/page page-id)] + [(seq record-key) (codec/encode-page keychain {:record-key record-key :page page})])) + pages)))] + (snapshot/consistent-view {:read-values (fn [record-keys] (mapv #(get records (seq %)) record-keys)) + :keychain keychain}))) -(deftest ^:integration snapshot-reads-its-revision-metadata-and-pages - (with-seeded-snapshot - (fn [raw-snapshot keychain] - (let [read-values #(ffi/await (ffi/read-snapshot-values raw-snapshot %)) - database-snapshot (snapshot/snapshot {:read-values read-values :revision-id "r-01K002" :keychain keychain}) - pages (revisions/fetch-pages-by-ids database-snapshot - {:ids #{1 2 3}})] - (is (= "r-01K002" (revisions/revision-id database-snapshot))) - (is (= 2 (revisions/size database-snapshot))) - (is (Arrays/equals (byte-array [2]) (get pages 1))) - (is (Arrays/equals (byte-array [9]) (get pages 2))) - (is (nil? (get pages 3))))))) +(defn- page-values [snapshot ids] + (update-vals (revisions/fetch-pages-by-ids snapshot {:ids ids}) seq)) -(deftest ^:integration reads-head-revision-id - (with-seeded-snapshot - (fn [raw-snapshot _keychain] - (let [read-values #(ffi/await (ffi/read-snapshot-values raw-snapshot %))] - (is (= "r-01K002" (snapshot/head-revision-id read-values))))))) +(deftest head-does-not-depend-on-history-and-hides-physical-tail-pages + (let [view (view {:head (head "r-head" 1) + :pages {1 (byte-array [7]) 2 (byte-array [9])}}) + current (revisions/head view)] + (is (= "r-head" (revisions/revision-id current))) + (is (= 1 (revisions/size current))) + (is (= {1 '(7) 2 nil} (page-values current #{1 2}))) + (is (= {1 '(7)} (page-values (revisions/as-of view "r-head") #{1}))) + (is (= #{} (revisions/changes-since view {:target-revision-id "r-head" :client-revision-id "r-head"}))))) -(deftest ^:integration consistent-view-reconstructs-an-as-of-snapshot-and-changes-since-it - (let [keychain (crypto-core/new-keychain) - object-store (ffi/open-object-store "memory:///") - database (ffi/open-database! object-store (str "d-" (UUID/randomUUID))) - root {:revision-id "r-root" - :parent nil - :database-page-count 1 - :pages {1 (byte-array [1])}} - first {:revision-id "r-first" - :parent "r-root" - :database-page-count 2 - :pages {2 (byte-array [2])}} - second {:revision-id "r-second" - :parent "r-first" - :database-page-count 2 - :pages {1 (byte-array [3])}} - records (mapcat (fn [revision] - (let [revision-key (key/revision (:revision-id revision))] - (concat [[revision-key - (codec/encode-revision keychain {:record-key revision-key :revision revision})]] - (map (fn [[page-id page]] - (let [page-key (key/page page-id)] - [page-key - (codec/encode-page keychain {:record-key page-key :page page})])) - (:pages revision))))) - [root first second])] - (try - (let [transaction (ffi/await (ffi/begin-transaction database))] - (try - (ffi/await - (ffi/write-values transaction - (concat records - [[(key/head) - (codec/encode-head {:revision-id "r-second"})]]))) - (ffi/await (ffi/commit-transaction transaction)) - (finally - (ffi/close-transaction! transaction)))) - (let [raw-snapshot (ffi/await (ffi/open-snapshot database))] - (try - (let [read-values #(ffi/await (ffi/read-snapshot-values raw-snapshot %)) - view (snapshot/consistent-view {:read-values read-values :keychain keychain}) - as-of (revisions/as-of view "r-first") - pages (revisions/fetch-pages-by-ids as-of {:ids #{1 2}})] - (is (= "r-first" (revisions/revision-id as-of))) - (is (= 2 (revisions/size as-of))) - (is (Arrays/equals (byte-array [1]) (get pages 1))) - (is (Arrays/equals (byte-array [2]) (get pages 2))) - (is (= #{1 2} - (revisions/changes-since view {:target-revision-id "r-second" :client-revision-id "r-root"})))) - (finally - (ffi/close-snapshot! raw-snapshot)))) - (finally - (ffi/close-database! database) - (ffi/close-object-store! object-store))))) +(deftest historical-reads-apply-every-newer-undo-and-stop-before-the-target + (let [view (view {:head (head "r-third" 2) + :pages {1 (byte-array [4]) 2 (byte-array [2])} + :history {"r-third" {:parent (head "r-second" 2) :undo-pages {1 (byte-array [3])}} + "r-second" {:parent (head "r-first" 2) :undo-pages {1 (byte-array [1])}} + "r-first" {:parent (head "r-root" 1) :undo-pages {2 nil}}}})] + (doseq [[revision-id size expected] + [["r-third" 2 {1 '(4) 2 '(2)}] + ["r-second" 2 {1 '(3) 2 '(2)}] + ["r-first" 2 {1 '(1) 2 '(2)}] + ["r-root" 1 {1 '(1) 2 nil}]]] + (let [snapshot (revisions/as-of view revision-id)] + (is (= size (revisions/size snapshot))) + (is (= expected (page-values snapshot #{1 2}))) + (is (= (select-keys expected [1]) (page-values snapshot #{1}))))) + (is (= #{1 2} (revisions/changes-since view {:target-revision-id "r-third" :client-revision-id "r-root"}))) + (is (= #{2} (revisions/changes-since view {:target-revision-id "r-first" :client-revision-id "r-root"}))))) -(deftest as-of-rejects-a-revision-with-an-incomplete-page-state - (let [keychain (crypto-core/new-keychain) - revision {:revision-id "r-incomplete" - :parent nil - :database-page-count 2 - :pages {1 (byte-array [1])}} - revision-key (key/revision (:revision-id revision)) - values {(seq revision-key) (codec/encode-revision keychain {:record-key revision-key :revision revision})} - read-values (fn [record-keys] - (mapv #(values (seq %)) record-keys)) - view (snapshot/consistent-view {:read-values read-values :keychain keychain}) - as-of (revisions/as-of view "r-incomplete")] - (try - (revisions/fetch-pages-by-ids as-of {:ids #{2}}) - (is false "as-of must reject incomplete page state") - (catch clojure.lang.ExceptionInfo error - (is (= "Revision does not contain page state" (.getMessage error))))))) +(deftest snapshots-reject-missing-in-range-pages + (let [view (view {:head (head "r-head" 2) :pages {1 (byte-array [1])}})] + (is (= ::revisions/incomplete-revision + (:reason (catch-ex-data #(page-values (revisions/head view) #{2}))))))) (deftest revision-chain-cycles-are-rejected - (let [keychain (crypto-core/new-keychain) - revisions [{:revision-id "r-cycle-a" :parent "r-cycle-b" - :database-page-count 1 :pages {}} - {:revision-id "r-cycle-b" :parent "r-cycle-a" - :database-page-count 1 :pages {}}] - values (into {} - (map (fn [revision] - (let [revision-key (key/revision (:revision-id revision))] - [(seq revision-key) - (codec/encode-revision keychain {:record-key revision-key :revision revision})])) - revisions)) - read-values (fn [record-keys] - (mapv #(values (seq %)) record-keys)) - view (snapshot/consistent-view {:read-values read-values :keychain keychain})] - (try - (revisions/changes-since view {:target-revision-id "r-cycle-a" :client-revision-id "r-missing"}) - (is false "revision cycles must be rejected") - (catch clojure.lang.ExceptionInfo error - (is (= "Revision chain contains a cycle" (.getMessage error))))))) + (let [view (view {:head (head "r-a" 1) + :history {"r-a" {:parent (head "r-b" 1) :undo-pages {}} + "r-b" {:parent (head "r-a" 1) :undo-pages {}}}})] + (doseq [operation [#(revisions/as-of view "r-missing") + #(revisions/changes-since view {:target-revision-id "r-a" :client-revision-id "r-missing"})]] + (is (= ::revisions/revision-cycle (:reason (catch-ex-data operation))))))) -(deftest missing-revisions-and-invalid-cursors-retain-their-domain-reasons - (let [keychain (crypto-core/new-keychain) - revision {:revision-id "r-head" - :parent nil - :database-page-count 1 - :pages {}} - revision-key (key/revision (:revision-id revision)) - values {(seq revision-key) (codec/encode-revision keychain {:record-key revision-key :revision revision})} - read-values (fn [record-keys] - (mapv #(values (seq %)) record-keys)) - view (snapshot/consistent-view {:read-values read-values :keychain keychain}) - reason (fn [f] - (try - (f) - (catch clojure.lang.ExceptionInfo error - (:reason (ex-data error)))))] +(deftest expired-history-and-unexpired-holes-have-distinct-outcomes + (doseq [[expires-at category reason expired?] + [[0 ::anomaly/not-found ::revisions/revision-not-found true] + [Long/MAX_VALUE ::anomaly/fault ::revisions/incomplete-revision false]]] + (let [view (view {:head (assoc (head "r-head" 1) :history-expires-at expires-at)}) + error (catch-ex-data #(revisions/as-of view "r-old"))] + (is (= category (::anomaly/category error))) + (is (= reason (:reason error))) + (is (= expired? (:expired? error)))))) + +(deftest missing-targets-and-invalid-cursors-retain-their-domain-reasons + (let [view (view {:head (head "r-head" 1) + :history {"r-head" {:parent nil :undo-pages {1 nil}}}})] (is (= ::revisions/revision-not-found - (reason #(revisions/as-of view "r-missing")))) + (:reason (catch-ex-data #(revisions/as-of view "r-missing"))))) (is (= ::revisions/invalid-revision-cursor - (reason #(revisions/changes-since view {:target-revision-id "r-head" :client-revision-id "r-missing"})))) - (is (= ::revisions/head-not-found - (reason #(snapshot/head-revision-id (constantly [nil]))))))) + (:reason (catch-ex-data #(revisions/changes-since view {:target-revision-id "r-head" :client-revision-id "r-missing"})))))) + (is (= ::revisions/head-not-found + (:reason (catch-ex-data #(snapshot/head-metadata (constantly [nil]))))))) + +(deftest unreadable-undo-is-not-treated-as-expired-history + (let [keychain (crypto-core/new-keychain) + values {(seq (key/head)) (codec/encode-head (assoc (head "r-head" 1) :history-expires-at 0)) + (seq (key/revision "r-head")) (byte-array [0])} + view (snapshot/consistent-view {:read-values (fn [record-keys] (mapv #(get values (seq %)) record-keys)) + :keychain keychain})] + (is (= ::codec/unreadable-record + (:reason (catch-ex-data #(revisions/as-of view "r-old"))))))) diff --git a/test/linear/adapter/slatedb/store_test.clj b/test/linear/adapter/slatedb/store_test.clj index 6f4c497..7bd5754 100644 --- a/test/linear/adapter/slatedb/store_test.clj +++ b/test/linear/adapter/slatedb/store_test.clj @@ -8,7 +8,9 @@ [linear.adapter.slatedb.ffi :as ffi] [linear.adapter.slatedb.key :as key] [linear.adapter.slatedb.store] - [linear.usecase.database.revisions :as revisions]) + [linear.usecase.database.model :as model] + [linear.usecase.database.revisions :as revisions] + [linear.usecase.keychain :as keychain]) (:import (java.util Arrays))) @@ -19,24 +21,8 @@ :created #inst "2026-01-01" :keychain keychain}}) -(defn- revision-records [keychain revision] - (let [revision-id (:revision-id revision) - revision-key (key/revision revision-id)] - (into [[revision-key - (codec/encode-revision keychain {:record-key revision-key :revision revision})]] - (concat - (map (fn [[page-id page]] - (let [page-key (key/page page-id)] - [page-key (codec/encode-page keychain {:record-key page-key :page page})])) - (:pages revision)) - [[(key/head) (codec/encode-head {:revision-id revision-id})]])))) - (defn- seed! [store keychain revision] - (with-open [database (connection/database store {:database-id "d-1"}) - transaction (connection/writable-transaction database)] - (ffi/await (ffi/write-values transaction - (revision-records keychain revision))) - (ffi/await (ffi/commit-transaction transaction)))) + (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"}) @@ -54,7 +40,7 @@ (try (is (= root (revisions/initialize! store {:database database :revision root}))) (let [[head-value page-value] (read-records store [(key/head) (key/page 1)])] - (is (= {:revision-id "r-created"} (codec/decode-head head-value))) + (is (= {:revision-id "r-created" :database-page-count 1} (select-keys (codec/decode-head head-value) [:revision-id :database-page-count]))) (is (not (Arrays/equals ^bytes (get-in root [:pages 1]) ^bytes page-value))) (is (Arrays/equals ^bytes (get-in root [:pages 1]) ^bytes (codec/decode-page keychain {:record-key (key/page 1) :value page-value})))) @@ -64,7 +50,7 @@ (catch clojure.lang.ExceptionInfo error error))] (is (= ::revisions/revision-conflict (:reason (ex-data error))))) (let [[head-value overwritten] (read-records store [(key/head) (key/revision "r-overwrite")])] - (is (= {:revision-id "r-created"} (codec/decode-head head-value))) + (is (= {:revision-id "r-created" :database-page-count 1} (select-keys (codec/decode-head head-value) [:revision-id :database-page-count]))) (is (nil? overwritten))) (finally (connection/close store))))) @@ -84,7 +70,7 @@ (catch clojure.lang.ExceptionInfo error error))] (is (= ::revisions/revision-conflict (:reason (ex-data error))))) (let [[head-value root-value] (read-records store [(key/head) (key/revision "r-created")])] - (is (= {} (codec/decode-head head-value))) + (is (= {} (select-keys (codec/decode-head head-value) [:revision-id :database-page-count]))) (is (nil? root-value))) (finally (connection/close store))))) @@ -129,10 +115,11 @@ (read-records store [(key/head) (key/revision "r-next") (key/page 1)])] - (is (= {:revision-id "r-next"} (codec/decode-head head-value))) - (is (= (dissoc revision :pages) - (dissoc (codec/decode-revision keychain {:record-key (key/revision "r-next") :value revision-value}) - :pages))) + (is (= {:revision-id "r-next" :database-page-count 1} (select-keys (codec/decode-head head-value) [:revision-id :database-page-count]))) + (let [undo (codec/decode-revision keychain {:record-key (key/revision "r-next") :value revision-value})] + (is (= "r-root" (get-in undo [:parent :revision-id]))) + (is (= 1 (get-in undo [:parent :database-page-count]))) + (is (Arrays/equals (byte-array [1]) (get-in undo [:undo-pages 1])))) (is (Arrays/equals (byte-array [2]) (codec/decode-page keychain {:record-key (key/page 1) :value page-value})))) (finally @@ -163,7 +150,7 @@ (read-records store [(key/head) (key/revision "r-stale") (key/page 1)])] - (is (= {:revision-id "r-root"} (codec/decode-head head-value))) + (is (= {:revision-id "r-root" :database-page-count 1} (select-keys (codec/decode-head head-value) [:revision-id :database-page-count]))) (is (nil? revision-value)) (is (Arrays/equals (byte-array [1]) (codec/decode-page keychain {:record-key (key/page 1) :value page-value})))) @@ -255,3 +242,134 @@ (when (and @borrowed (zero? @returned)) ((:return-database store) "d-1" @borrowed)) (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))))) + +(defn- record-expiries [store record-keys] + (with-open [database (connection/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))) + +(deftest ^:integration history-expires-but-current-data-survives-inactivity-and-remains-writable + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + keychain (crypto-core/new-keychain) + database (database-record keychain) + root {:revision-id "r-root" :parent nil :database-page-count 2 + :pages {1 (byte-array [1]) 2 (byte-array [2])}} + next {:revision-id "r-next" :parent "r-root" :database-page-count 2 + :pages {1 (byte-array [3])}} + before (System/currentTimeMillis)] + (try + (seed! store keychain root) + (let [[head-expiry page-expiry revision-expiry] + (record-expiries store [(key/head) (key/page 1) (key/revision "r-root")])] + (is (nil? head-expiry)) + (is (nil? page-expiry)) + (is (<= (+ before (* 30 86400000)) revision-expiry))) + (expire-history! store ["r-root"]) + (revisions/with-consistent-view [view store database] + (let [snapshot (revisions/head view)] + (is (= 2 (revisions/size snapshot))) + (is (= {1 '(1) 2 '(2)} + (update-vals (revisions/fetch-pages-by-ids snapshot {:ids #{1 2}}) seq))))) + (revisions/publish-next! store {:revision next :database database}) + (revisions/with-consistent-view [view store database] + (let [snapshot (revisions/as-of view "r-root")] + (is (= 2 (revisions/size snapshot))) + (is (= {1 '(1) 2 '(2)} + (update-vals (revisions/fetch-pages-by-ids snapshot {:ids #{1 2}}) seq))))) + (expire-history! store ["r-next"]) + (revisions/with-consistent-view [view store database] + (is (= {1 '(3) 2 '(2)} + (update-vals (revisions/fetch-pages-by-ids (revisions/head view) {:ids #{1 2}}) seq)))) + (finally (connection/close store))))) + +(deftest ^:integration shorter-retention-does-not-create-a-hole-in-the-undo-chain + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1 :revision-retention-days 90}) + keychain (crypto-core/new-keychain) + database (database-record keychain) + root {:revision-id "r-root" :parent nil :database-page-count 1 :pages {1 (byte-array [1])}}] + (try + (seed! store keychain root) + (revisions/publish-next! (assoc store :revision-retention-days 1) + {:database database :revision {:revision-id "r-next" :parent "r-root" :database-page-count 1 :pages {1 (byte-array [2])}}}) + (let [[root-expiry next-expiry] (record-expiries store [(key/revision "r-root") (key/revision "r-next")])] + (is (= root-expiry next-expiry)) + (is (<= (+ (System/currentTimeMillis) (* 89 86400000)) next-expiry))) + (finally (connection/close store))))) + +(deftest ^:integration shrink-regrow-and-repeated-page-updates-preserve-every-state + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + keychain (crypto-core/new-keychain) + database (database-record keychain) + root {:revision-id "r-root" :parent nil :database-page-count 2 + :pages {1 (byte-array [1]) 2 (byte-array [2])}}] + (try + (seed! store keychain root) + (doseq [revision [{:revision-id "r-shrink" :parent "r-root" :database-page-count 1 :pages {1 (byte-array [3])}} + {:revision-id "r-grow" :parent "r-shrink" :database-page-count 3 :pages {2 (byte-array [4]) 3 (byte-array [5])}} + {:revision-id "r-repeat" :parent "r-grow" :database-page-count 3 :pages {1 (byte-array [6])}}]] + (revisions/publish-next! store {:database database :revision revision})) + (revisions/with-consistent-view [view store database] + (doseq [[id expected] [["r-root" {1 '(1) 2 '(2) 3 nil}] + ["r-shrink" {1 '(3) 2 nil 3 nil}] + ["r-grow" {1 '(3) 2 '(4) 3 '(5)}] + ["r-repeat" {1 '(6) 2 '(4) 3 '(5)}]]] + (is (= expected + (update-vals (revisions/fetch-pages-by-ids (revisions/as-of view id) {:ids #{1 2 3}}) seq))))) + (finally (connection/close store))))) + +(deftest ^:integration failed-page-encryption-rolls-back-the-staged-undo-and-allows-retry + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + keychain (crypto-core/new-keychain) + database (database-record keychain) + failing-keychain (reify keychain/Keychain + (-encrypt [_ value options] + (when (Arrays/equals ^bytes (key/page 2) ^bytes (:associated-data options)) + (throw (ex-info "Page encryption failed" {}))) + (keychain/-encrypt keychain value options)) + (-decrypt [_ value options] + (keychain/-decrypt keychain value options))) + root {:revision-id "r-root" :parent nil :database-page-count 1 :pages {1 (byte-array [1])}} + next {:revision-id "r-next" :parent "r-root" :database-page-count 2 + :pages (sorted-map 1 (byte-array [2]) 2 (byte-array [3]))}] + (try + (seed! store keychain root) + (is (thrown? clojure.lang.ExceptionInfo + (revisions/publish-next! store {:database (assoc-in database [:vault :keychain] failing-keychain) :revision next}))) + (let [[head-value undo-value new-page] (read-records store [(key/head) (key/revision "r-next") (key/page 2)])] + (is (= "r-root" (:revision-id (codec/decode-head head-value)))) + (is (nil? undo-value)) + (is (nil? new-page))) + (is (= next (revisions/publish-next! store {:database database :revision next}))) + (revisions/with-consistent-view [view store database] + (is (= '(1) (seq (get (revisions/fetch-pages-by-ids (revisions/as-of view "r-root") {:ids #{1}}) 1))))) + (finally (connection/close store))))) + +(deftest ^:integration expired-client-cursors-bootstrap-the-fixed-target-but-missing-targets-fail + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + keychain (crypto-core/new-keychain) + database (database-record keychain)] + (try + (seed! store keychain {:revision-id "r-root" :parent nil :database-page-count 1 :pages {1 (byte-array [1])}}) + (revisions/publish-next! store {:database database :revision {:revision-id "r-first" :parent "r-root" :database-page-count 1 :pages {1 (byte-array [2])}}}) + (expire-history! store ["r-root" "r-first"]) + (revisions/publish-next! store {:database database :revision {:revision-id "r-second" :parent "r-first" :database-page-count 2 :pages {2 (byte-array [3])}}}) + (revisions/with-consistent-view [view store database] + (let [database (model/database database {::model/consistent-view view}) + result (model/pull database {:server-revision "r-first" :client-revision "r-root"})] + (is (= "r-first" (:server-revision result))) + (is (= 1 (:database-page-count result))) + (is (= {1 '(2)} (update-vals (:pages result) seq))) + (is (thrown? clojure.lang.ExceptionInfo (model/pull database {:server-revision "r-root"}))))) + (finally (connection/close store))))) diff --git a/test/linear/handler/turso/protobuf_test.clj b/test/linear/handler/turso/protobuf_test.clj index bfd22fb..00fb3dc 100644 --- a/test/linear/handler/turso/protobuf_test.clj +++ b/test/linear/handler/turso/protobuf_test.clj @@ -2,8 +2,13 @@ (:require [clojure.test :refer [deftest is testing]] [cognitect.anomalies :as anomaly] + [linear.adapter.crypto.core :as crypto] + [linear.adapter.slatedb.connection :as connection] + [linear.adapter.slatedb.store] [linear.handler.turso.protobuf :as protobuf] [linear.test :as test] + [linear.usecase.database.model :as model] + [linear.usecase.database.revisions :as revisions] [ring.core.protocols :as ring]) (:import (java.io ByteArrayOutputStream DataOutputStream) @@ -179,3 +184,34 @@ (let [data (test/catch-ex-data decode)] (is (= ::anomaly/incorrect (::anomaly/category data))) (is (string? (::anomaly/message data)))))) + +(deftest ^:integration incremental-pulls-across-shrinkage-stream-only-existing-target-pages + (let [store (connection/open {:object-store-url "memory:///" :max-open-databases 1}) + database {:id "d-shrink" :display-name "Shrink" + :vault {:id "v-shrink" :created java.time.Instant/EPOCH :keychain (crypto/new-keychain)}} + assert-pull (fn [options] + (revisions/with-consistent-view [view store database] + (let [pull (model/pull (model/database database {::model/consistent-view view}) options) + expected-pages (if (= #{2} (:page-ids options)) {} {1 '(3)})] + (is (= "r-shrink" (:server-revision pull))) + (is (= 1 (:database-page-count pull))) + (is (= expected-pages (update-vals (:pages pull) seq))) + (with-open [output (ByteArrayOutputStream.)] + (ring/write-body-to-stream (protobuf/pull-stream pull) nil output) + (is (pos? (.size output)))))))] + (try + (revisions/initialize! store + {:database database + :revision {:revision-id "r-root" :parent nil :database-page-count 1 :pages {1 (byte-array [1])}}}) + (doseq [revision [{:revision-id "r-grow" :parent "r-root" :database-page-count 2 :pages {2 (byte-array [2])}} + {:revision-id "r-shrink" :parent "r-grow" :database-page-count 1 :pages {1 (byte-array [3])}}]] + (revisions/publish-next! store {:database database :revision revision})) + (doseq [options [{:client-revision "r-root"} + {:client-revision "r-root" :page-ids #{1 2}} + {:client-revision "r-root" :page-ids #{2}}]] + (assert-pull options)) + (revisions/publish-next! store + {:database database + :revision {:revision-id "r-regrow" :parent "r-shrink" :database-page-count 2 :pages {2 (byte-array [4])}}}) + (assert-pull {:client-revision "r-root" :server-revision "r-shrink"}) + (finally (connection/close store))))) diff --git a/test/linear/test_data/slatedb.clj b/test/linear/test_data/slatedb.clj index e122118..57a0648 100644 --- a/test/linear/test_data/slatedb.clj +++ b/test/linear/test_data/slatedb.clj @@ -1,30 +1,13 @@ (ns linear.test-data.slatedb (:require - [linear.adapter.slatedb.codec :as codec] - [linear.adapter.slatedb.connection :as connection] - [linear.adapter.slatedb.ffi :as ffi] - [linear.adapter.slatedb.key :as key])) + [linear.adapter.slatedb.store] + [linear.usecase.database.revisions :as revisions])) -(defn root-records [{:keys [keychain pages revision-id parent] - :or {revision-id "r-root" - parent nil}}] - (let [revision {:revision-id revision-id - :parent parent - :database-page-count (count pages) - :pages pages} - revision-key (key/revision (:revision-id revision))] - (into [[revision-key - (codec/encode-revision keychain {:record-key revision-key :revision revision})]] - (concat - (map (fn [[page-id page]] - (let [page-key (key/page page-id)] - [page-key (codec/encode-page keychain {:record-key page-key :page page})])) - pages) - [[(key/head) - (codec/encode-head {:revision-id (:revision-id revision)})]])))) - -(defn store-root! [{:keys [store database-id] :as options}] - (with-open [database (connection/database store {:database-id database-id}) - transaction (connection/writable-transaction database)] - (ffi/await (ffi/write-values transaction (root-records options))) - (ffi/await (ffi/commit-transaction transaction)))) +(defn store-root! [{:keys [store database-id keychain pages revision-id] + :or {revision-id "r-root"}}] + (revisions/initialize! store + {:database {:id database-id :vault {:keychain keychain}} + :revision {:revision-id revision-id + :parent nil + :database-page-count (count pages) + :pages pages}})) diff --git a/test/linear/usecase/database/lifecycle_test.clj b/test/linear/usecase/database/lifecycle_test.clj index b7f3e85..140e546 100644 --- a/test/linear/usecase/database/lifecycle_test.clj +++ b/test/linear/usecase/database/lifecycle_test.clj @@ -7,14 +7,18 @@ [linear.adapter.slatedb.store] [linear.adapter.sqlite.evaluator] [linear.test :refer [catch-ex-data run]] + [linear.test-data.sqlite :as sqlite-data] [linear.usecase.core :as core] [linear.usecase.database :as database] + [linear.usecase.database.evaluator :as evaluator] [linear.usecase.database.revisions :as revisions] [linear.usecase.grant :as grant] [linear.usecase.vault :as vault] [next.jdbc :as jdbc] [next.jdbc.result-set :as result-set]) (:import + (java.nio.file Files) + (java.sql DriverManager) (java.util Arrays))) (def ^:private components @@ -177,3 +181,38 @@ (vault/delete! ctx {:actor "manager" :vault-id source-vault}) (is (= ::anomaly/conflict (::anomaly/category (catch-ex-data #(database/create! ctx (assoc (create-args target-vault) :source {:database-id source-id})))))))))) + +(defn- auto-vacuum-pages [] + (let [path (Files/createTempFile "linear-auto-vacuum" ".db" + (make-array java.nio.file.attribute.FileAttribute 0))] + (try + (with-open [connection (DriverManager/getConnection (str "jdbc:sqlite:" path)) + statement (.createStatement connection)] + (.execute statement "PRAGMA auto_vacuum=FULL") + (.execute statement "PRAGMA journal_mode=WAL") + (.execute statement "CREATE TABLE payload (id INTEGER PRIMARY KEY, data BLOB)")) + (sqlite-data/pages {:image (Files/readAllBytes path)}) + (finally (Files/deleteIfExists path))))) + +(deftest ^:integration sqlite-shrink-and-regrowth-preserve-historical-page-images + (with-system [system (run {:keys components})] + (let [ctx (context system) + pages (auto-vacuum-pages) + initializer (reify evaluator/Initializer + (-initial-revision [_] + {:revision-id "r-auto-vacuum" :parent nil :database-page-count (count pages) :pages pages})) + id (database/create! (assoc ctx ::core/evaluator initializer) (create-args (new-vault ctx "manager"))) + args {:actor "manager" :database-id id} + fill-sql (str "WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM n WHERE x<64) " + "INSERT INTO payload SELECT x, zeroblob(4096) FROM n") + push (fn [sqls] (database/push! ctx (assoc args :command {:statements (mapv #(hash-map :sql % :parameters []) sqls)})))] + (push [fill-sql]) + (let [large (database/pull ctx args)] + (push ["DELETE FROM payload"]) + (let [small (database/pull ctx args)] + (is (< (:database-page-count small) (:database-page-count large))) + (push [fill-sql]) + (doseq [expected [large small]] + (let [actual (database/pull ctx (assoc args :server-revision (:server-revision expected)))] + (is (= (:database-page-count expected) (:database-page-count actual))) + (is (= (update-vals (:pages expected) seq) (update-vals (:pages actual) seq)))))))))) diff --git a/test/linear/usecase/database/model_test.clj b/test/linear/usecase/database/model_test.clj index 747d661..3267838 100644 --- a/test/linear/usecase/database/model_test.clj +++ b/test/linear/usecase/database/model_test.clj @@ -164,3 +164,53 @@ (is (not (spec/valid? input-schema (dissoc database capability)))))))) (is (not (spec/valid? ::revisions/consistent-view (model/database attributes {::model/consistent-view view})))))) + +(deftest expired-client-history-fetches-all-selected-pages-from-the-fixed-target + (let [fetched (atom nil) + target (snapshot "r-target" 3 (fn [{:keys [ids]}] (reset! fetched ids) {})) + view (reify revisions/ConsistentView + (-head [_] (throw (AssertionError. "must keep the explicit target"))) + (-as-of [_ revision-id] + (is (= "r-target" revision-id)) + target) + (-changes-since [_ target-id client-id] + (is (= ["r-target" "r-old"] [target-id client-id])) + (throw (ex-info "Expired" {::anomaly/category ::anomaly/not-found + :reason ::revisions/revision-not-found :expired? true})))) + database (model/database attributes {::model/consistent-view view})] + (is (= "r-target" (:server-revision (model/pull database {:client-revision "r-old" :server-revision "r-target" :page-ids #{2 3 4}})))) + (is (= #{2 3} @fetched)))) + +(deftest history-corruption-and-unavailable-explicit-targets-are-not-full-pull-fallbacks + (doseq [failure [{::anomaly/category ::anomaly/fault :reason ::revisions/incomplete-revision} + {::anomaly/category ::anomaly/fault :reason :linear.adapter.slatedb.codec/unreadable-record} + {::anomaly/category ::anomaly/fault :reason ::revisions/revision-cycle} + {::anomaly/category ::anomaly/not-found :reason ::revisions/revision-not-found}]] + (let [error (ex-info "History failed" failure) + current (snapshot "r-head" 1 (fn [_] (throw (AssertionError. "must not fetch")))) + view (reify revisions/ConsistentView + (-head [_] current) + (-as-of [_ _] (throw error)) + (-changes-since [_ _ _] (throw error))) + database (model/database attributes {::model/consistent-view view})] + (doseq [options [{:client-revision "r-old"} {:server-revision "r-missing"}]] + (is (identical? error (try (model/pull database options) + (catch clojure.lang.ExceptionInfo caught caught)))))))) + +(deftest pull-excludes-pages-outside-the-target-size-after-shrinkage + (doseq [[page-count options expected-ids] + [[1 {:client-revision "r-old"} #{1}] + [1 {:client-revision "r-old" :page-ids #{1 2}} #{1}] + [1 {:client-revision "r-old" :page-ids #{2}} #{}] + [1 {:client-revision "r-old" :server-revision "r-target"} #{1}] + [0 {:client-revision "r-old"} #{}]]] + (let [target (snapshot "r-target" page-count + (fn [{:keys [ids]}] + (is (= expected-ids ids)) + (zipmap ids (repeat (byte-array [1]))))) + head (snapshot "r-newer" 3 (fn [_] (throw (AssertionError. "must use the target snapshot")))) + view (view (if (:server-revision options) head target) {"r-target" target} #{1 2}) + result (model/pull (model/database attributes {::model/consistent-view view}) options)] + (is (= "r-target" (:server-revision result))) + (is (= page-count (:database-page-count result))) + (is (= expected-ids (set (keys (:pages result))))))))