feat(revisions): retain history with expiring undo records

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:

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

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

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 {}
 

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

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)))

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))