diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index 189029aa0f..529fcd1cc8 100644 --- a/.serena/memories/backend/storage.md +++ b/.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 diff --git a/backend/src/app/jobs/metrics.clj b/backend/src/app/jobs/metrics.clj index 5ba7df2864..ab1941b1b7 100644 --- a/backend/src/app/jobs/metrics.clj +++ b/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"]) diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 7476f9bfb2..2f119a769f 100644 --- a/backend/src/app/main.clj +++ b/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 diff --git a/backend/src/app/tasks/file_gc_scheduler.clj b/backend/src/app/tasks/file_gc_scheduler.clj index 7eaa3801cc..4ac26acf6e 100644 --- a/backend/src/app/tasks/file_gc_scheduler.clj +++ b/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)))) diff --git a/backend/src/app/tasks/offload_file_data.clj b/backend/src/app/tasks/offload_file_data.clj index fae8b2d75d..e8e3eb12ff 100644 --- a/backend/src/app/tasks/offload_file_data.clj +++ b/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] diff --git a/backend/test/backend_tests/file_gc_scheduler_test.clj b/backend/test/backend_tests/file_gc_scheduler_test.clj index 0edd217acb..a8702d44ef 100644 --- a/backend/test/backend_tests/file_gc_scheduler_test.clj +++ b/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"}))))))) diff --git a/backend/test/backend_tests/offload_file_data_test.clj b/backend/test/backend_tests/offload_file_data_test.clj new file mode 100644 index 0000000000..f4c99c3310 --- /dev/null +++ b/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)))))))) diff --git a/docs/technical-guide/configuration.md b/docs/technical-guide/configuration.md index b239d53551..aa4593fa61 100644 --- a/docs/technical-guide/configuration.md +++ b/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.