✨ Harden file data offload job with lock, summary and metrics (#12403)

Take the file row with FOR UPDATE SKIP LOCKED when the offload job
starts and skip the run when the row is locked or gone, so the job
never runs while the file is being edited. Every live hot-storage
row of the file moves to cold storage inside one all-or-nothing
transaction, and the handler answers a summary map with the
offloaded rows, bytes and per-type stats.

Add bounded metrics for the job: rows and bytes counters labeled by
row type on successful runs only, and a timing histogram with the
ok, skip and error outcomes. The job definition now requires
metrics, the recorders live in app.jobs.metrics, and the existing
ephemeral job timing histogram pins its buckets.

Cover the job with a new test namespace: the happy path, the locked
row, missing and deleted files, the rows and bytes summary with its
no-op rerun, the whole-file abort on storage failure, the metric
definitions and emissions, and the submit-time params validation.
An end-to-end suite reads, snapshots, writes, exports, imports and
deletes offloaded files; concurrency tests pin the stale-revn skip,
the blocked-edit retry and the reclaimable orphans of a partial
failure. The file-gc scheduler test pins one job per eligible file,
and the scheduler private helper drops its bang.

Document the lock invariant, the deliberate no-pacing decision and
the metric label sets in the backend storage memory and the
configuration guide.

Related to #12398.

AI-assisted-by: muse-spark-1.3-contributor

.serena/memories/backend/storage.md

@@ -125,6 +125,11 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
 - The `file_data.metadata.storage-ref-id` value points to the storage object.
 - `fdata/upsert!` touches a storage object from incoming metadata before it stores the new row.
 - File snapshots use `file_data` for snapshot data and `file_change` for snapshot metadata.
+- Tiered offload (`:tiered-file-data-storage` flag + `db` backend): `file-gc` submits an `:offload-file-data` job (priority 10, delay 1000, same params) only when it actually processed the file; the handler takes the `file` row with `FOR UPDATE SKIP LOCKED` and skips (`{:skipped 1}`, no error) when the row is locked or gone, then moves every live `file_data` row of that file with `backend = 'db'` to `backend = 'storage'` via `fdata/upsert!` (one `put-object!` in bucket `file-data` per row, `data` becomes nil) inside a single all-or-nothing transaction per file. The skip only serializes because every `file_data` writer holds at least `FOR SHARE` on the `file` row. A storage failure aborts only that file and bubbles for a runner retry; orphans go to `storage-gc-touched`. A second run is a no-op because the query only matches `backend = 'db'`.
+- Offload pacing: none by deliberate decision. Sleeps only make sense when batching multiple transactions; the per-file transaction is short by construction and the scheduler run takes minutes per production metrics, so neither loop sleeps. `jobs/heartbeat` (pool-based liveness, ~1 write/s) stays on both paths.
+- Offload metrics (bounded labels only, never ids or error text): `penpot_file_data_offload_rows_total{type}` and `penpot_file_data_offload_bytes_total{type}` (`type`: `main`/`snapshot`/`fragment`, successful runs only) plus `penpot_file_data_offload_timing_ms{result}` (`result`: `ok`/`skip`/`error`). Helpers live in `app.jobs.metrics`; the job-def requires `::mtx/metrics`.
+- Writes always resolve the backend through `fdata/default-backend` (`cf/get :file-data-backend`), so post-offload edits land back in hot (`db`) storage without extra code; reads stay transparent through `fdata/resolve-file-data` + `decode-file-data` + `realize`.
+- Deleting an offloaded row touches its storage object first (`objects-gc/delete-file-data-chunk` via `fdata/process-metadata`), so `storage-gc-touched` reclaims the blob with the usual delay; restoring a file unmarks the referenced objects.
 
 ## Metrics
 

backend/src/app/jobs/metrics.clj

@@ -176,6 +176,46 @@
     (record-value metrics :jobs-request-timing [outcome]
                   (max 0 (long millis)))))
 
+(defn- offload-result-label
+  "Coerce an offload outcome to its label: lowercase, with `failed`
+  only when there is no outcome at all. Values come from the fixed
+  call sites of the offload handler (`ok`, `skip`, `error`), never
+  from params, IDs or exception text."
+  [result]
+  (str/lower (mtx/label result "failed")))
+
+(defn- offload-type-label
+  "Coerce a file data type to its label: lowercase, with `other` only
+  when there is no type at all. Values are the fixed `file_data.type`
+  values (`main`, `snapshot`, `fragment`)."
+  [type]
+  (str/lower (mtx/label type "other")))
+
+(defn record-offload-rows
+  "Count file data rows moved by the offload job, broken down by row
+  type. Only successful runs emit rows, so there is no result label."
+  [cfg type amount]
+  (record-count (mtx/instance cfg) :file-data-offload-rows
+                [(offload-type-label type)]
+                amount))
+
+(defn record-offload-bytes
+  "Count file data bytes moved by the offload job, broken down by row
+  type. Only successful runs emit bytes, so there is no result label."
+  [cfg type amount]
+  (record-count (mtx/instance cfg) :file-data-offload-bytes
+                [(offload-type-label type)]
+                amount))
+
+(defn record-offload-duration
+  "Time one offload run from lock to summary (milliseconds). Recorded
+  for every outcome (`ok`, `skip`, `error`), so its count is the run
+  total per outcome."
+  [cfg result millis]
+  (record-value (mtx/instance cfg) :file-data-offload-timing
+                [(offload-result-label result)]
+                (max 0 (long millis))))
+
 (def ^:private backlog-statuses
   ["new" "scheduled" "running" "retry" "completed" "failed" "cancelled" "aborted"])
 

backend/src/app/main.clj

@@ -389,7 +389,26 @@
    {::mdef/name "penpot_audit_ingest_events_total"
     ::mdef/help "Audit log frontend events ingested, by kind and outcome."
     ::mdef/labels ["kind" "outcome"]
-    ::mdef/type :counter}})
+    ::mdef/type :counter}
+
+   :file-data-offload-rows
+   {::mdef/name "penpot_file_data_offload_rows_total"
+    ::mdef/help "File data rows moved from hot to cold storage by the offload job."
+    ::mdef/labels ["type"]
+    ::mdef/type :counter}
+
+   :file-data-offload-bytes
+   {::mdef/name "penpot_file_data_offload_bytes_total"
+    ::mdef/help "File data bytes moved from hot to cold storage by the offload job."
+    ::mdef/labels ["type"]
+    ::mdef/type :counter}
+
+   :file-data-offload-timing
+   {::mdef/name "penpot_file_data_offload_timing_ms"
+    ::mdef/help "Time spent offloading one file from hot to cold storage (milliseconds)."
+    ::mdef/labels ["result"]
+    ::mdef/type :histogram
+    ::mdef/buckets [1 10 50 100 500 1000 5000 10000 30000 60000 300000]}})
 
 (def system-config
   {::db/pool

backend/src/app/tasks/file_gc_scheduler.clj

@@ -29,7 +29,7 @@
       FOR UPDATE OF f
      SKIP LOCKED")
 
-(defn- schedule!
+(defn- schedule-files
   [{:keys [::db/conn] :as cfg} threshold]
   (let [total (reduce (fn [total {:keys [id modified-at revn]}]
                         (let [params {:file-id id :revn revn}]
@@ -83,4 +83,4 @@
                       (ct/in-past))]
     (-> cfg
         (assoc ::db/rollback (:rollback? params))
-        (db/tx-run! schedule! threshold))))
+        (db/tx-run! schedule-files threshold))))

backend/src/app/tasks/offload_file_data.clj

@@ -10,9 +10,12 @@
   (:require
    [app.common.logging :as l]
    [app.common.schema :as sm]
+   [app.common.time :as ct]
    [app.db :as db]
    [app.features.fdata :as fdata]
    [app.jobs :as jobs]
+   [app.jobs.metrics :as metrics]
+   [app.metrics :as mtx]
    [app.storage :as sto]
    [integrant.core :as ig]))
 
@@ -23,27 +26,83 @@
       AND fd.backend = 'db'
       AND fd.deleted_at IS NULL")
 
+(def ^:private sql:lock-file
+  ;; NOTE: SKIP LOCKED only serializes against writers that also lock
+  ;; the file row. Every file_data writer must therefore hold at least
+  ;; FOR SHARE on it (the update RPC via lock-for-share?, file-gc via
+  ;; FOR UPDATE); a writer skipping that lock would reopen the
+  ;; lost-update race this lock closes.
+  "SELECT id FROM file WHERE id = ? AND deleted_at IS NULL FOR UPDATE SKIP LOCKED")
+
 (defn- offload-file-data
-  [cfg {:keys [id file-id type] :as fdata}]
+  "Offload one hot-storage row to cold storage. Returns the row type
+  and the migrated bytes for the caller to aggregate."
+  [cfg {:keys [id file-id type data] :as fdata}]
   (jobs/heartbeat cfg)
   (fdata/upsert! cfg (assoc fdata :backend "storage"))
-  (l/trc :file-id (str file-id)
-         :id (str id)
-         :type type))
+  (let [size (if (bytes? data) (alength ^bytes data) 0)]
+    (l/trc :hint "file data row offloaded"
+           :file-id (str file-id)
+           :id (str id)
+           :type type
+           :bytes size)
+    {:type type
+     :size size}))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; HANDLER
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
+(defn- offload-file-rows
+  "Offload every live hot-storage row of one file inside the caller
+  transaction. Takes the file row lock first and skips when the row
+  is locked or gone. Returns a summary map with the offloaded rows,
+  bytes, per-type stats and the skipped flag."
+  [{:keys [::db/conn] :as cfg} file-id]
+  (if (db/exec-one! conn [sql:lock-file file-id])
+    (let [{:keys [rows bytes by-type]}
+          (reduce (fn [stats fdata]
+                    (let [{:keys [type size]} (offload-file-data cfg fdata)]
+                      (-> stats
+                          (update :rows inc)
+                          (update :bytes + size)
+                          (update-in [:by-type type]
+                                     (fn [{r :rows b :bytes :or {r 0 b 0}}]
+                                       {:rows (inc r) :bytes (+ b size)})))))
+                  {:rows 0 :bytes 0 :by-type {}}
+                  (db/plan conn [sql:get-file-data file-id]))]
+      (l/trc :hint "file data offloaded"
+             :file-id (str file-id)
+             :offloaded rows
+             :bytes bytes)
+      {:offloaded rows :bytes bytes :by-type by-type :skipped 0})
+    (do
+      (l/dbg :hint "skip offload, file is locked or gone"
+             :file-id (str file-id))
+      {:offloaded 0 :bytes 0 :by-type {} :skipped 1})))
+
 (defn execute-offload-file-data
   "Plain job handler: offload the file data rows of one file."
   [cfg params]
-  (let [file-id (:file-id params)]
-    (-> cfg
-        (assoc ::db/rollback (:rollback? params))
-        (db/tx-run! (fn [{:keys [::db/conn] :as cfg}]
-                      (run! (partial offload-file-data cfg)
-                            (db/plan conn [sql:get-file-data file-id])))))))
+  (let [file-id (:file-id params)
+        tpoint  (ct/tpoint)]
+    (try
+      (let [summary (-> cfg
+                        (assoc ::db/rollback (:rollback? params))
+                        (db/tx-run! offload-file-rows file-id))]
+        (if (pos? (:skipped summary))
+          (do
+            (metrics/record-offload-duration cfg :skip (inst-ms (tpoint)))
+            summary)
+          (do
+            (doseq [[type {:keys [rows bytes]}] (:by-type summary)]
+              (metrics/record-offload-rows cfg type rows)
+              (metrics/record-offload-bytes cfg type bytes))
+            (metrics/record-offload-duration cfg :ok (inst-ms (tpoint)))
+            summary)))
+      (catch Throwable cause
+        (metrics/record-offload-duration cfg :error (inst-ms (tpoint)))
+        (throw cause)))))
 
 (def schema:offload-file-data-params
   [:map
@@ -52,7 +111,8 @@
 (defmethod ig/assert-key ::job-def
   [_ params]
   (assert (db/pool? (::db/pool params)) "expected a valid database pool")
-  (assert (sto/valid-storage? (::sto/storage params)) "expected valid storage to be provided"))
+  (assert (sto/valid-storage? (::sto/storage params)) "expected valid storage to be provided")
+  (assert (some? (::mtx/metrics params)) "expected metrics to be provided"))
 
 (defmethod ig/init-key ::job-def
   [_ cfg]

backend/test/backend_tests/file_gc_scheduler_test.clj

@@ -6,6 +6,7 @@
 
 (ns backend-tests.file-gc-scheduler-test
   (:require
+   [app.common.time :as ct]
    [app.jobs :as jobs]
    [backend-tests.helpers :as th]
    [clojure.test :as t]))
@@ -26,3 +27,17 @@
       (t/is (thrown-with-msg? Exception #"check error"
                               (jobs/submit cfg {::jobs/name   :file-gc-scheduler
                                                 ::jobs/params {:min-age {:bogus true}}}))))))
+
+(t/deftest scheduler-submits-one-job-per-eligible-file
+  (let [profile (th/create-profile* 1)]
+    (doseq [i [1 2 3]]
+      (let [file (th/create-file* i {:profile-id (:id profile)
+                                     :project-id (:default-project-id profile)})]
+        ;; Make the files deterministically eligible: pending trim and
+        ;; modified long enough ago.
+        (th/db-update! :file {:has-media-trimmed false
+                              :modified-at (ct/in-past "2h")}
+                       {:id (:id file)})))
+    (let [result (th/run-task! :file-gc-scheduler {:min-age "1h"})]
+      (t/is (= 3 (:processed result)))
+      (t/is (= 3 (count (th/db-query :job {:name "file-gc"})))))))

backend/test/backend_tests/offload_file_data_test.clj

@@ -0,0 +1,611 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) KALEIDOS SUBSIDIARY SL
+
+(ns backend-tests.offload-file-data-test
+  (:require
+   [app.binfile.common :as bfc]
+   [app.binfile.v3 :as v3]
+   [app.common.features :as cfeat]
+   [app.common.time :as ct]
+   [app.common.types.shape :as cts]
+   [app.common.uuid :as uuid]
+   [app.config :as cf]
+   [app.db :as db]
+   [app.features.fdata :as fdata]
+   [app.jobs :as jobs]
+   [app.jobs.metrics :as metrics]
+   [app.main :as main]
+   [app.metrics :as mtx]
+   [app.metrics.definition :as-alias mdef]
+   [app.rpc :as-alias rpc]
+   [app.storage :as sto]
+   [app.storage.tmp :as tmp]
+   [app.tasks.offload-file-data :as offload]
+   [backend-tests.helpers :as th]
+   [clojure.test :as t]
+   [datoteka.io :as io]
+   [integrant.core :as ig])
+  (:import
+   io.prometheus.client.Counter
+   io.prometheus.client.Counter$Child))
+
+(t/use-fixtures :once th/state-init)
+(t/use-fixtures :each th/database-reset)
+
+(defn- db-rows
+  [file-id]
+  (th/db-query :file-data {:file-id file-id}))
+
+(defn- create-file
+  [i profile]
+  (th/create-file* i {:profile-id (:id profile)
+                      :project-id (:default-project-id profile)
+                      :is-shared false}))
+
+(t/deftest offloads-file-data-to-storage
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        before  (db-rows (:id file))]
+
+    (t/is (pos? (count before)))
+    (t/is (every? #(= "db" (:backend %)) before))
+
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (= (count before) (:offloaded result)))
+      (t/is (zero? (:skipped result))))
+
+    (let [after (db-rows (:id file))]
+      (t/is (= (count before) (count after)))
+      (t/is (every? #(= "storage" (:backend %)) after))
+      (t/is (every? #(nil? (:data %)) after)))))
+
+(t/deftest skips-file-locked-by-another-transaction
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        before  (db-rows (:id file))
+        locked  (promise)
+        release (promise)
+        holder  (future
+                  (db/tx-run! th/*system*
+                              (fn [{:keys [::db/conn]}]
+                                (db/exec-one! conn ["SELECT id FROM file WHERE id = ? FOR UPDATE" (:id file)])
+                                (deliver locked true)
+                                @release
+                                :held)))]
+
+    (t/is (= true (deref locked 10000 ::timeout)))
+
+    (try
+      (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+        (t/is (= 0 (:offloaded result)))
+        (t/is (= 1 (:skipped result))))
+      (finally
+        (deliver release true)
+        (deref holder 10000 ::timeout)))
+
+    (let [after (db-rows (:id file))]
+      (t/is (= (count before) (count after)))
+      (t/is (every? #(= "db" (:backend %)) after)))))
+
+(t/deftest skips-missing-and-deleted-files
+  (let [result (th/run-task! :offload-file-data {:file-id (uuid/next)})]
+    (t/is (= 0 (:offloaded result)))
+    (t/is (= 1 (:skipped result))))
+
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        _       (th/db-update! :file {:deleted-at (ct/now)} {:id (:id file)})
+        before  (db-rows (:id file))
+        result  (th/run-task! :offload-file-data {:file-id (:id file)})]
+    (t/is (= 0 (:offloaded result)))
+    (t/is (= 1 (:skipped result)))
+    (let [after (db-rows (:id file))]
+      (t/is (= (count before) (count after)))
+      (t/is (every? #(= "db" (:backend %)) after)))))
+
+(defn- row-bytes
+  [{:keys [data]}]
+  (if (bytes? data)
+    (alength ^bytes data)
+    0))
+
+(t/deftest summary-reports-rows-and-bytes
+  (let [profile        (th/create-profile* 1)
+        file           (create-file 1 profile)
+        before         (db-rows (:id file))
+        expected-bytes (reduce + 0 (map row-bytes before))]
+
+    (t/is (pos? (count before)))
+    (t/is (every? some? (map :data before)))
+    (t/is (pos? expected-bytes))
+
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (= (count before) (:offloaded result)))
+      (t/is (= expected-bytes (:bytes result)))
+      (t/is (zero? (:skipped result))))
+
+    ;; A second run is a no-op: nothing left on the hot storage.
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (= 0 (:offloaded result)))
+      (t/is (= 0 (:bytes result)))
+      (t/is (zero? (:skipped result))))))
+
+(t/deftest storage-failure-aborts-only-that-file
+  (let [profile  (th/create-profile* 1)
+        file-a   (create-file 1 profile)
+        file-b   (create-file 2 profile)
+        real-put @#'sto/put-object!
+        failing  (fn [storage params]
+                   (if (= (:file-id params) (:id file-a))
+                     (throw (ex-info "storage is down" {:file-id (:id file-a)}))
+                     (real-put storage params)))]
+
+    ;; The failing file aborts as a whole: every row stays on hot storage.
+    (with-redefs [sto/put-object! failing]
+      (t/is (thrown? Exception
+                     (th/run-task! :offload-file-data {:file-id (:id file-a)}))))
+
+    (let [rows-a (db-rows (:id file-a))]
+      (t/is (pos? (count rows-a)))
+      (t/is (every? #(= "db" (:backend %)) rows-a))
+      (t/is (every? some? (map :data rows-a))))
+
+    ;; The other file is independent and offloads while storage is down for A.
+    (with-redefs [sto/put-object! failing]
+      (let [result (th/run-task! :offload-file-data {:file-id (:id file-b)})]
+        (t/is (pos? (:offloaded result)))
+        (t/is (pos? (:bytes result)))
+        (t/is (zero? (:skipped result)))))
+
+    (t/is (every? #(= "storage" (:backend %)) (db-rows (:id file-b))))
+
+    ;; Once storage recovers, a retry converges with no leftovers on hot.
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file-a)})]
+      (t/is (pos? (:offloaded result)))
+      (t/is (pos? (:bytes result)))
+      (t/is (zero? (:skipped result))))
+
+    (let [rows-a (db-rows (:id file-a))]
+      (t/is (every? #(= "storage" (:backend %)) rows-a))
+      (t/is (every? #(nil? (:data %)) rows-a)))))
+
+(defn- make-metrics
+  []
+  (ig/init-key :app.metrics/metrics
+               {:default (select-keys main/default-metrics
+                                      [:file-data-offload-rows
+                                       :file-data-offload-bytes
+                                       :file-data-offload-timing])}))
+
+(defn- offload-cfg
+  [metrics]
+  (-> th/*system*
+      (assoc ::mtx/metrics metrics)
+      (assoc ::sto/storage (assoc (::sto/storage th/*system*)
+                                  ::mtx/metrics metrics))))
+
+(defn- counter-value
+  [metrics id labels]
+  (let [collector (mtx/get-collector metrics id)
+        instance  (::mdef/instance collector)
+        child     (.labels ^Counter instance (into-array String labels))]
+    (.get ^Counter$Child child)))
+
+(def ^:private known-types
+  ["main" "snapshot" "fragment"])
+
+(defn- total-counter
+  [metrics id]
+  (reduce + 0 (map #(counter-value metrics id [%]) known-types)))
+
+(t/deftest metric-definitions-are-pinned
+  (let [defs main/default-metrics]
+    (t/is (= "penpot_file_data_offload_rows_total"
+             (get-in defs [:file-data-offload-rows ::mdef/name])))
+    (t/is (= ["type"]
+             (get-in defs [:file-data-offload-rows ::mdef/labels])))
+    (t/is (= :counter
+             (get-in defs [:file-data-offload-rows ::mdef/type])))
+    (t/is (= "penpot_file_data_offload_bytes_total"
+             (get-in defs [:file-data-offload-bytes ::mdef/name])))
+    (t/is (= ["type"]
+             (get-in defs [:file-data-offload-bytes ::mdef/labels])))
+    (t/is (= :counter
+             (get-in defs [:file-data-offload-bytes ::mdef/type])))
+    (t/is (= "penpot_file_data_offload_timing_ms"
+             (get-in defs [:file-data-offload-timing ::mdef/name])))
+    (t/is (= ["result"]
+             (get-in defs [:file-data-offload-timing ::mdef/labels])))
+    (t/is (= :histogram
+             (get-in defs [:file-data-offload-timing ::mdef/type])))
+    (t/is (= [1 10 50 100 500 1000 5000 10000 30000 60000 300000]
+             (get-in defs [:file-data-offload-timing ::mdef/buckets])))))
+
+(t/deftest success-emits-rows-and-bytes
+  (let [metrics (make-metrics)
+        profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        result  (offload/execute-offload-file-data (offload-cfg metrics)
+                                                   {:file-id (:id file)})]
+    (t/is (pos? (:offloaded result)))
+    (t/is (= (double (:offloaded result))
+             (total-counter metrics :file-data-offload-rows)))
+    (t/is (= (double (:bytes result))
+             (total-counter metrics :file-data-offload-bytes)))
+    (t/is (pos? (counter-value metrics :file-data-offload-rows ["main"])))
+    (t/is (pos? (total-counter metrics :file-data-offload-rows)))))
+
+(t/deftest skip-records-skip-and-nothing-else
+  (let [durations (volatile! [])
+        rows      (volatile! [])
+        bytes     (volatile! [])]
+    (with-redefs [metrics/record-offload-duration (fn [_ result _] (vswap! durations conj result) nil)
+                  metrics/record-offload-rows (fn [_ type amount] (vswap! rows conj [type amount]) nil)
+                  metrics/record-offload-bytes (fn [_ type amount] (vswap! bytes conj [type amount]) nil)]
+      (let [result (offload/execute-offload-file-data th/*system* {:file-id (uuid/next)})]
+        (t/is (= 0 (:offloaded result)))
+        (t/is (= 1 (:skipped result)))))
+    (t/is (= [:skip] @durations))
+    (t/is (empty? @rows))
+    (t/is (empty? @bytes))))
+
+(t/deftest error-records-error-and-rethrows
+  (let [profile  (th/create-profile* 1)
+        file     (create-file 1 profile)
+        durations (volatile! [])
+        real-put @#'sto/put-object!]
+    (with-redefs [sto/put-object! (fn [_ _] (throw (ex-info "storage is down" {})))
+                  metrics/record-offload-duration (fn [_ result _] (vswap! durations conj result) nil)]
+      (t/is (thrown? Exception
+                     (offload/execute-offload-file-data th/*system* {:file-id (:id file)}))))
+    (t/is (= [:error] @durations))
+    ;; Nothing was migrated: the whole file aborted.
+    (t/is (every? #(= "db" (:backend %)) (db-rows (:id file))))))
+
+(t/deftest missing-metrics-raises
+  (let [params {::db/pool th/*pool*
+                ::sto/storage (::sto/storage th/*system*)}]
+    (t/is (thrown? AssertionError
+                   (ig/assert-key :app.tasks.offload-file-data/job-def params)))
+    (t/is (nil? (ig/assert-key :app.tasks.offload-file-data/job-def
+                               (assoc params ::mtx/metrics (:app.metrics/metrics th/*system*)))))))
+
+(t/deftest submit-validates-params-upfront
+  (let [cfg th/*system*]
+    (t/testing "valid params submit a durable job"
+      (t/is (uuid? (jobs/submit cfg {::jobs/name   :offload-file-data
+                                     ::jobs/params {:file-id (uuid/next)}}))))
+    (t/testing "invalid params fail at submit instead of burning retries"
+      (t/is (thrown? Exception
+                     (jobs/submit cfg {::jobs/name   :offload-file-data
+                                       ::jobs/params {}})))
+      (t/is (thrown? Exception
+                     (jobs/submit cfg {::jobs/name   :offload-file-data
+                                       ::jobs/params {:file-id "not-a-uuid"}}))))))
+
+(t/deftest file-gc-without-flag-does-not-enqueue-offload
+  (t/is (not (contains? cf/flags :tiered-file-data-storage)))
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)]
+    (t/is (true? (th/run-task! :file-gc {:file-id (:id file)})))
+    (th/run-pending-jobs)
+    (t/testing "no offload job was enqueued"
+      (t/is (empty? (th/db-query :job {:name "offload-file-data"}))))
+    (t/testing "every row stays on hot storage"
+      (let [rows (db-rows (:id file))]
+        (t/is (pos? (count rows)))
+        (t/is (every? #(= "db" (:backend %)) rows))
+        (t/is (every? some? (map :data rows)))))))
+
+(defn- update-file!
+  [& {:keys [profile-id file-id changes revn] :or {revn 0}}]
+  (let [params {::th/type :update-file
+                ::rpc/profile-id profile-id
+                :id file-id
+                :session-id (uuid/random)
+                :revn revn
+                :vern 0
+                :features cfeat/supported-features
+                :changes changes}
+        out    (th/command! params)]
+    (t/is (nil? (:error out)))
+    (:result out)))
+
+(defn- add-page-with-rect
+  [profile file page-id shape-id]
+  (update-file!
+   :file-id (:id file)
+   :profile-id (:id profile)
+   :changes
+   [{:type :add-page
+     :name "test"
+     :id page-id}
+    {:type :add-obj
+     :page-id page-id
+     :id shape-id
+     :parent-id uuid/zero
+     :frame-id uuid/zero
+     :components-v2 true
+     :obj (cts/setup-shape
+           {:id shape-id
+            :name "rect"
+            :frame-id uuid/zero
+            :parent-id uuid/zero
+            :type :rect})}]))
+
+(defn- get-file-data
+  [profile file-id]
+  (let [out (th/command! {::th/type :get-file
+                          ::rpc/profile-id (:id profile)
+                          :id file-id})]
+    (t/is (nil? (:error out)))
+    (get-in out [:result :data])))
+
+(t/deftest offloaded-file-reads-identically
+  (let [profile  (th/create-profile* 1 {:is-active true})
+        file     (create-file 1 profile)
+        page-id  (uuid/next)
+        shape-id (uuid/next)]
+    (add-page-with-rect profile file page-id shape-id)
+
+    (let [before-bfc (:data (bfc/get-file th/*system* (:id file) {:realize? true}))
+          before-rpc (get-file-data profile (:id file))]
+
+      (t/is (some? (get-in before-bfc [:pages-index page-id :objects shape-id])))
+      (t/is (= before-bfc before-rpc))
+
+      (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+        (t/is (pos? (:offloaded result)))
+        (t/is (every? #(= "storage" (:backend %)) (remove :deleted-at (db-rows (:id file))))))
+
+      (t/testing "direct read resolves cold storage transparently"
+        (t/is (= before-bfc
+                 (:data (bfc/get-file th/*system* (:id file) {:realize? true})))))
+
+      (t/testing "rpc read resolves cold storage transparently"
+        (t/is (= before-rpc (get-file-data profile (:id file))))))))
+
+(t/deftest snapshot-survives-offload
+  (let [profile  (th/create-profile* 1 {:is-active true})
+        file     (create-file 1 profile)
+        page-id  (uuid/next)
+        shape-id (uuid/next)]
+    (add-page-with-rect profile file page-id shape-id)
+
+    (let [out (th/command! {::th/type :create-file-snapshot
+                            ::rpc/profile-id (:id profile)
+                            :file-id (:id file)
+                            :label "before-offload"})]
+      (t/is (nil? (:error out)))
+      (t/is (uuid? (get-in out [:result :id]))))
+
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (pos? (:offloaded result))))
+
+    (t/testing "live snapshot rows moved to cold storage"
+      (let [snapshots (th/db-query :file-data {:file-id (:id file)
+                                               :type "snapshot"
+                                               :deleted-at nil})]
+        (t/is (pos? (count snapshots)))
+        (t/is (every? #(= "storage" (:backend %)) snapshots))
+        (t/is (every? #(nil? (:data %)) snapshots))))
+
+    (t/testing "snapshots are still listed"
+      (let [out (th/command! {::th/type :get-file-snapshots
+                              ::rpc/profile-id (:id profile)
+                              :file-id (:id file)})]
+        (t/is (nil? (:error out)))
+        (t/is (some #(= "before-offload" (:label %)) (:result out)))))
+
+    (t/testing "snapshot restores from cold storage"
+      (let [snap-id (->> (:result (th/command! {::th/type :get-file-snapshots
+                                                ::rpc/profile-id (:id profile)
+                                                :file-id (:id file)}))
+                         (filter #(= "before-offload" (:label %)))
+                         first
+                         :id)
+            out     (th/command! {::th/type :restore-file-snapshot
+                                  ::rpc/profile-id (:id profile)
+                                  :file-id (:id file)
+                                  :id snap-id})]
+        (t/is (some? snap-id))
+        (t/is (nil? (:error out)))
+        (t/is (some? (get-in (get-file-data profile (:id file))
+                             [:pages-index page-id :objects shape-id])))))))
+
+(t/deftest write-after-offload-returns-to-hot-storage
+  (let [profile  (th/create-profile* 1 {:is-active true})
+        file     (create-file 1 profile)
+        page-id  (uuid/next)
+        shape-id (uuid/next)]
+
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (pos? (:offloaded result))))
+
+    (let [main (th/db-get :file-data {:file-id (:id file) :type "main"})]
+      (t/is (= "storage" (:backend main)))
+      (t/is (nil? (:data main))))
+
+    (add-page-with-rect profile file page-id shape-id)
+
+    (t/testing "the new main row is back on hot storage"
+      (let [main (th/db-get :file-data {:file-id (:id file) :type "main"})]
+        (t/is (= "db" (:backend main)))
+        (t/is (bytes? (:data main)))))
+
+    (t/testing "the written content reads back"
+      (t/is (some? (get-in (get-file-data profile (:id file))
+                           [:pages-index page-id :objects shape-id]))))))
+
+(t/deftest export-import-roundtrip-after-offload
+  (let [profile  (th/create-profile* 1 {:is-active true})
+        file     (create-file 1 profile)
+        page-id  (uuid/next)
+        shape-id (uuid/next)]
+    (add-page-with-rect profile file page-id shape-id)
+
+    (let [before (:data (bfc/get-file th/*system* (:id file) {:realize? true}))
+          _      (th/run-task! :offload-file-data {:file-id (:id file)})
+          output (tmp/tempfile :suffix ".zip")]
+
+      (v3/export-files!
+       (-> th/*system*
+           (assoc ::bfc/ids #{(:id file)})
+           (assoc ::bfc/export-type :detach-libraries))
+       (io/output-stream output))
+
+      (let [result   (-> th/*system*
+                         (assoc ::bfc/project-id (:default-project-id profile))
+                         (assoc ::bfc/profile-id (:id profile))
+                         (assoc ::bfc/team-id (:default-team-id profile))
+                         (assoc ::bfc/input output)
+                         (v3/import-files!))
+            imported (:data (bfc/get-file th/*system* (first (:file-ids result))
+                                          {:realize? true}))]
+        (t/is (= 1 (count (:file-ids result))))
+        (t/is (= (set (keys (:pages-index before)))
+                 (set (keys (:pages-index imported)))))
+        (t/is (= (get-in before [:pages-index page-id :name])
+                 (get-in imported [:pages-index page-id :name])))
+        (t/is (some? (get-in imported [:pages-index page-id :objects shape-id])))))))
+
+(t/deftest deleted-offloaded-file-is-fully-purged
+  (let [storage (sto/resolve th/*system*)
+        profile (th/create-profile* 1 {:is-active true})
+        file    (create-file 1 profile)
+        page-id (uuid/next)]
+
+    (add-page-with-rect profile file page-id (uuid/next))
+
+    (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+      (t/is (pos? (:offloaded result))))
+
+    (let [ref-ids (into #{}
+                        (comp (map :metadata)
+                              (map fdata/decode-metadata)
+                              (keep :storage-ref-id))
+                        (db-rows (:id file)))]
+      (t/is (pos? (count ref-ids)))
+      (t/is (every? some? (map #(sto/get-object storage %) ref-ids)))
+
+      (let [out (th/command! {::th/type :delete-file
+                              ::rpc/profile-id (:id profile)
+                              :id (:id file)})]
+        (t/is (nil? (:error out))))
+      (th/run-pending-jobs)
+
+      (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
+                  (th/run-task! :objects-gc {}))]
+        (t/is (pos? (:processed res))))
+
+      (t/testing "every file data row is gone"
+        (t/is (empty? (db-rows (:id file)))))
+
+      ;; The touches above landed 8 days in the future, so the touched
+      ;; collector needs a clock past them plus its deletion delay.
+      (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 9}))]
+                  (th/run-task! :storage-gc-touched {}))]
+        (t/is (pos? (:delete res))))
+
+      (t/testing "every referenced storage object is reclaimed"
+        (let [objects (mapv #(sto/get-object storage %) ref-ids)]
+          (t/is (every? nil? objects)))))))
+
+(t/deftest file-gc-with-stale-revn-enqueues-nothing
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        before  (db-rows (:id file))
+        stale   (inc (or (:revn file) 0))]
+    (t/is (not (th/run-task! :file-gc {:file-id (:id file) :revn stale})))
+    (th/run-pending-jobs)
+    (t/testing "no offload job was enqueued"
+      (t/is (empty? (th/db-query :job {:name "offload-file-data"}))))
+    (t/testing "every row stays on hot storage"
+      (let [after (db-rows (:id file))]
+        (t/is (= (count before) (count after)))
+        (t/is (every? #(= "db" (:backend %)) after))))))
+
+(t/deftest concurrent-edit-and-offload-stay-coherent
+  (let [profile  (th/create-profile* 1 {:is-active true})
+        file     (create-file 1 profile)
+        page-1   (uuid/next)
+        shape-1  (uuid/next)
+        page-2   (uuid/next)
+        shape-2  (uuid/next)]
+    (add-page-with-rect profile file page-1 shape-1)
+
+    (let [locked  (promise)
+          release (promise)
+          holder  (future
+                    (db/tx-run! th/*system*
+                                (fn [{:keys [::db/conn]}]
+                                  (db/exec-one! conn ["SELECT id FROM file WHERE id = ? FOR UPDATE" (:id file)])
+                                  (deliver locked true)
+                                  @release
+                                  :held)))
+          _       (t/is (= true (deref locked 10000 ::timeout)))
+          updater (future
+                    (add-page-with-rect profile file page-2 shape-2))]
+
+      (try
+        ;; Give the updater time to block on the held row lock.
+        (th/sleep 500)
+        (t/is (not (realized? updater)))
+
+        (t/testing "offload skips while the row is locked"
+          (let [result (th/run-task! :offload-file-data {:file-id (:id file)})]
+            (t/is (= 0 (:offloaded result)))
+            (t/is (= 1 (:skipped result)))))
+        (finally
+          (deliver release true)))
+
+      (t/is (not= ::timeout (deref holder 10000 ::timeout)))
+      (t/is (not= ::timeout (deref updater 30000 ::timeout))))
+
+    (t/testing "the blocked edit won as a whole"
+      (t/is (some? (get-in (get-file-data profile (:id file))
+                           [:pages-index page-2 :objects shape-2]))))
+
+    (t/testing "a later offload migrates the new bytes, never stale ones"
+      (let [current (get-file-data profile (:id file))
+            result  (th/run-task! :offload-file-data {:file-id (:id file)})]
+        (t/is (pos? (:offloaded result)))
+        (t/is (= current (get-file-data profile (:id file))))))))
+
+(t/deftest partial-failure-leaves-reclaimable-orphans
+  (let [profile (th/create-profile* 1)
+        file    (create-file 1 profile)
+        _       (t/is (>= (count (db-rows (:id file))) 2))
+        calls   (volatile! 0)
+        orphans (volatile! [])
+        real-put @#'sto/put-object!
+        mock    (fn [storage params]
+                  (if (= (:file-id params) (:id file))
+                    (let [n (vswap! calls inc)]
+                      (if (= 1 n)
+                        (let [object (real-put storage params)]
+                          (vswap! orphans conj (:id object))
+                          object)
+                        (throw (ex-info "storage is down" {}))))
+                    (real-put storage params)))]
+
+    (with-redefs [sto/put-object! mock]
+      (t/is (thrown? Exception
+                     (th/run-task! :offload-file-data {:file-id (:id file)}))))
+
+    (t/testing "the aborted file keeps every row on hot storage"
+      (let [rows (db-rows (:id file))]
+        (t/is (every? #(= "db" (:backend %)) rows))
+        (t/is (every? some? (map :data rows)))))
+
+    (t/testing "the orphaned blob is reclaimed once old enough"
+      (t/is (= 1 (count @orphans)))
+      (let [storage (sto/resolve th/*system*)]
+        (t/is (some? (sto/get-object storage (first @orphans))))
+        (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
+                    (th/run-task! :storage-gc-touched {}))]
+          (t/is (pos? (:delete res))))
+        (t/is (nil? (sto/get-object storage (first @orphans))))))))

docs/technical-guide/configuration.md

@@ -828,6 +828,8 @@ PENPOT_FLAGS: [...] enable-tiered-file-data-storage
 PENPOT_FILE_DATA_BACKEND: db
 ```
 
+Once enabled, the offload job moves one file per run inside a single transaction: it locks the file row (skipping the run while the file is being edited) and migrates every live data row to the objects storage. You can follow it with `penpot_file_data_offload_rows_total` and `penpot_file_data_offload_bytes_total` (label `type="main|snapshot|fragment"`) and `penpot_file_data_offload_timing_ms` (label `result="ok|skip|error"`).
+
 ### Autosave
 
 By default, Penpot stores manually saved versions indefinitely; these can be found in the History tab and can be renamed, restored, deleted, etc. Additionally, the default behavior of on-premise instances is to not keep automatic version history. This automatic behavior can be modified and adapted to each on-premise installation with the corresponding configuration.