From 7993c53c77f2d8a13cad0a37930a6e7fd2497db4 Mon Sep 17 00:00:00 2001 From: Haokang Den Date: Fri, 23 Oct 2026 00:05:54 +0900 Subject: [PATCH 1/2] feat(revisions)!: retain history with expiring undo records Forward page deltas depend on ancestors beyond the retention window. Capture before-images with their parent state in the publish transaction so SlateDB TTL and compaction can reclaim history independently. Keep HEAD and current pages without expiry so inactivity never removes the current database. Retain replaced states for 30 configurable days, preserve expiry ordering when retention is shortened, and bootstrap expired client cursors against the selected target. BREAKING CHANGE: HEAD and stored revisions use a new format. No migration is provided because the application has not been released. --- compose.yaml | 1 + docs/architecture.md | 15 ++ docs/deployment.md | 7 + duct.edn | 6 +- src/linear/adapter/slatedb/connection.clj | 11 +- src/linear/adapter/slatedb/ffi.clj | 17 +- src/linear/adapter/slatedb/snapshot.clj | 204 +++++++-------- src/linear/adapter/slatedb/store.clj | 50 +++- src/linear/usecase/database/model.clj | 18 +- test/linear/adapter/slatedb/codec_test.clj | 17 +- test/linear/adapter/slatedb/ffi_test.clj | 49 +++- test/linear/adapter/slatedb/snapshot_test.clj | 242 ++++++------------ test/linear/adapter/slatedb/store_test.clj | 170 ++++++++++-- test/linear/test_data/slatedb.clj | 37 +-- .../usecase/database/lifecycle_test.clj | 39 +++ test/linear/usecase/database/model_test.clj | 32 +++ 16 files changed, 552 insertions(+), 363 deletions(-) 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..1e7e8cf 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,10 +109,7 @@ (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})) + changed-page-ids (changed-page-ids view target-revision client-revision page-count) selected-page-ids (if page-ids (set (filter page-ids changed-page-ids)) changed-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/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..030b2ec 100644 --- a/test/linear/usecase/database/model_test.clj +++ b/test/linear/usecase/database/model_test.clj @@ -164,3 +164,35 @@ (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)))))))) From d7173e676b86880d9a4d979ec5e494390fbb92db Mon Sep 17 00:00:00 2001 From: Haokang Den Date: Fri, 23 Oct 2026 08:30:51 +0900 Subject: [PATCH 2/2] fix(sync): omit truncated pages from incremental pulls A cursor spanning page growth followed by shrinkage can include removed tail pages in its change set. Reading those pages produces nil values that cannot be serialized by the protobuf pull stream. Bound selected page IDs to the target snapshot size before fetching them. Cover page selectors, empty targets, and historical targets after HEAD has regrown, including the real SlateDB-to-protobuf streaming path. --- src/linear/usecase/database/model.clj | 7 ++-- test/linear/handler/turso/protobuf_test.clj | 36 +++++++++++++++++++++ test/linear/usecase/database/model_test.clj | 18 +++++++++++ 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/src/linear/usecase/database/model.clj b/src/linear/usecase/database/model.clj index 1e7e8cf..d12483a 100644 --- a/src/linear/usecase/database/model.clj +++ b/src/linear/usecase/database/model.clj @@ -110,9 +110,10 @@ target-revision (revisions/revision-id snapshot) page-count (revisions/size snapshot) changed-page-ids (changed-page-ids view target-revision client-revision page-count) - selected-page-ids (if page-ids - (set (filter page-ids changed-page-ids)) - changed-page-ids)] + 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/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/usecase/database/model_test.clj b/test/linear/usecase/database/model_test.clj index 030b2ec..3267838 100644 --- a/test/linear/usecase/database/model_test.clj +++ b/test/linear/usecase/database/model_test.clj @@ -196,3 +196,21 @@ (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))))))))