diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index 88a0c30042..0e2abd696b 100644 --- a/.serena/memories/backend/storage.md +++ b/.serena/memories/backend/storage.md @@ -25,7 +25,7 @@ - Existing object values can remain readable until physical deletion. - `:expired-at` blocks reads after the expiration time. - `del-object!` sets `deleted_at`. It does not remove backend content. -- `storage-gc-deleted` removes the database row and backend content after the deletion delay, streaming the pending rows with `db/plan` inside a read-only transaction and deleting them in chunks of 25, each chunk in its own transaction: a chunk that fails (for example, an object a foreign key still references) is logged and skipped instead of aborting the run. +- `storage-gc-deleted` removes the database row and backend content after the deletion delay, processing the pending objects in chunks of 25 with keyset pagination (`(deleted_at, id)` cursor, matching the `storage_object(deleted_at, id)` index). Each chunk is one short read-write transaction: `SELECT ... FOR UPDATE SKIP LOCKED` (so two concurrent runs never take the same chunk; no separate read transaction exists and the run does not block the table vacuum) followed by the database delete and the backend delete grouped by backend. The cursor advances to the `(deleted_at, id)` of the last row read regardless of the deletion result, so a chunk that fails is skipped and not visited again in the same run: the error aborts the transaction (nothing of the chunk is deleted, locks are released), the handler catches it inside the transaction function and rolls it back explicitly — the commit that `tx-run!` performs afterwards is a JDBC-driver no-op on the already-rolled-back connection — and the loop continues with the returned rows. Accepted quirks, both benign and self-healing: (1) with rows of several backends in one chunk, a failure after one group's physical delete leaves those rows pending with already-removed content — pending rows are excluded from reads and the next run finishes them, since the physical delete is idempotent (S3 ignores missing keys, FS uses `deleteIfExists`); (2) rows locked by a concurrent run are skipped via `SKIP LOCKED` and deferred to the next scheduled run, and the `deleted_at <= now` cutoff is sampled per chunk. It emits the `penpot_storage_gc_deleted_objects_total{backend}` counter (one increment per deleted backend group, reported only after the chunk commits, so rolled-back chunks are never counted twice); run duration is covered by the generic `penpot_tasks_timing{name="storage-gc-deleted"}` histogram and chunk failures by the log-derived `penpot_log_events_total`. Every 5000 deleted objects the run pauses 5 seconds (throttle between chunk transactions, never holding one open). - `storage-gc-touched` finds references before it sets `deleted_at`. - `objects-gc` removes deleted domain rows and touches their storage object IDs. Each proc streams its pending keys with `db/plan` inside a read-only transaction and deletes them in chunks, each chunk in its own transaction with `FOR UPDATE SKIP LOCKED`: a chunk that fails is logged and skipped instead of aborting the run. - Use `::db/reuse-conn true` with `sto/resolve` inside a database transaction. diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 221c20af3a..65e55047fe 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -105,6 +105,12 @@ ::mdef/labels ["name"] ::mdef/type :histogram} + :storage-gc-deleted-objects + {::mdef/name "penpot_storage_gc_deleted_objects_total" + ::mdef/help "Storage objects permanently deleted by storage-gc-deleted, by backend." + ::mdef/labels ["backend"] + ::mdef/type :counter} + :redis-eval-timing {::mdef/name "penpot_redis_eval_timing" ::mdef/help "Redis EVAL commands execution timings (ms)" @@ -230,7 +236,8 @@ ::sto.gc-deleted/handler {::db/pool (ig/ref ::db/pool) - ::sto/storage (ig/ref ::sto/storage)} + ::sto/storage (ig/ref ::sto/storage) + ::mtx/metrics (ig/ref ::mtx/metrics)} ::sto.gc-touched/handler {::db/pool (ig/ref ::db/pool)} diff --git a/backend/src/app/storage/gc_deleted.clj b/backend/src/app/storage/gc_deleted.clj index 8356e35b0e..197889a3fb 100644 --- a/backend/src/app/storage/gc_deleted.clj +++ b/backend/src/app/storage/gc_deleted.clj @@ -2,7 +2,7 @@ ;; 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 INC Sucursal en España SL +;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.storage.gc-deleted "A task responsible to permanently delete already marked as deleted @@ -17,110 +17,164 @@ [app.common.logging :as l] [app.common.time :as ct] [app.db :as db] + [app.metrics :as mtx] [app.storage :as sto] [app.storage.impl :as impl] [integrant.core :as ig])) -(def ^:private sql:lock-sobjects - "SELECT id FROM storage_object - WHERE id = ANY(?::uuid[]) - FOR UPDATE - SKIP LOCKED") - -(defn- lock-ids - "Perform a select before delete for proper object locking and - prevent concurrent operations and we proceed only with successfully - locked objects." - [conn ids] - (let [ids (db/create-array conn "uuid" ids)] - (->> (db/exec! conn [sql:lock-sobjects ids]) - (into #{} (map :id)) - (not-empty)))) - (def ^:private sql:delete-sobjects "DELETE FROM storage_object WHERE id = ANY(?::uuid[])") -(defn- delete-sobjects - [conn ids] - (let [ids (db/create-array conn "uuid" ids)] - (-> (db/exec-one! conn [sql:delete-sobjects ids]) - (db/get-update-count)))) - -(defn- delete-in-bulk - [cfg backend-id ids] - ;; Each chunk runs in its own transaction, on its own connection: a chunk - ;; that fails (for example, an object that a foreign key still references) is - ;; logged and skipped, and it cannot poison the cursor that feeds the rest. - (try - (db/tx-run! cfg - (fn [{:keys [::db/conn ::sto/storage]}] - (when-let [ids (lock-ids conn ids)] - (let [total (delete-sobjects conn ids)] - (-> (impl/resolve-backend storage backend-id) - (impl/del-objects-in-bulk ids)) - - (doseq [id ids] - (l/dbg :hint "permanently delete storage object" - :id (str id) - :backend (name backend-id))) - total)))) - (catch Throwable cause - (l/err :hint "unexpected error on bulk deletion" - :ids ids - :cause cause)))) +(defn- delete-by-backend + "Deletes the locked objects of one backend group: the database rows and + their backend content. Runs inside the caller's transaction. Returns the + number of deleted rows; reporting it (metrics) is the caller's job, once + the transaction has committed." + [{:keys [::db/conn ::sto/storage]} backend-id ids] + (let [ids-array (db/create-array conn "uuid" ids) + total (-> (db/exec-one! conn [sql:delete-sobjects ids-array]) + (db/get-update-count))] + (-> (impl/resolve-backend storage backend-id) + (impl/del-objects-in-bulk ids)) + (doseq [id ids] + (l/dbg :hint "permanently delete storage object" + :id (str id) + :backend (name backend-id))) + total)) (defn- group-by-backend - [items] - (d/group-by (comp keyword :backend) :id #{} items)) + [rows] + (d/group-by (comp keyword :backend) :id #{} rows)) + +(defn- process-chunk-deletion + "Deletes the locked rows of one chunk, grouped by backend. Runs inside the + caller's transaction: any failure here aborts it, so the whole chunk is + rolled back (nothing is deleted and the locks are released) and skipped by + the caller. Returns a map of backend id to deleted row count." + [tx rows] + (->> (group-by-backend rows) + (map (fn [[backend-id ids]] + [backend-id (delete-by-backend tx backend-id ids)])) + (into {}))) (def ^:private sql:get-deleted-sobjects - "SELECT s.* + "SELECT s.id, s.backend, s.deleted_at FROM storage_object AS s WHERE s.deleted_at IS NOT NULL AND s.deleted_at <= ? - ORDER BY s.deleted_at ASC") + AND (s.deleted_at, s.id) > (?, ?) + ORDER BY s.deleted_at ASC, s.id ASC + LIMIT ? + FOR UPDATE SKIP LOCKED") + +(def ^:private sql:get-first-deleted-sobjects + "SELECT s.id, s.backend, s.deleted_at + FROM storage_object AS s + WHERE s.deleted_at IS NOT NULL + AND s.deleted_at <= ? + ORDER BY s.deleted_at ASC, s.id ASC + LIMIT ? + FOR UPDATE SKIP LOCKED") (def ^:private chunk-size - "Pending objects deleted in each transaction." + "Pending objects read, locked and deleted in each transaction." 25) -(defn- pending-row - "Takes the columns the GC needs from a `db/plan` row. +(def ^:private throttle-every + "Deleted objects between throttle pauses." + 5000) - A plan row is a view over the open result set, valid only while the reduction - is running, so it cannot be stored for later: this has to happen inside the - reducing function." - [row] - {:id (:id row) - :backend (:backend row)}) +(def ^:private throttle-ms + "Pause applied every `throttle-every` deleted objects, letting the + database (and its replicas) breathe on huge backlogs." + 5000) + +(defn- throttle + "Sleeps `throttle-ms` when the deleted count crosses a `throttle-every` + boundary. Runs between chunk transactions, never holding one open." + [prev-total new-total] + (when (> (quot new-total throttle-every) + (quot prev-total throttle-every)) + (l/inf :hint "throttling storage gc" + :deleted new-total + :pause-ms throttle-ms) + (Thread/sleep (long throttle-ms)))) + +(defn- fetch-and-lock-chunk + "Reads and locks one chunk of pending objects in `(deleted_at, id)` order. + Objects already locked by another process are skipped, so two concurrent + runs never take the same chunk. Runs inside the caller's transaction." + [conn last-deleted-at last-id] + (let [[sql params] (if (some? last-deleted-at) + [sql:get-deleted-sobjects + [(ct/now) last-deleted-at last-id chunk-size]] + [sql:get-first-deleted-sobjects + [(ct/now) chunk-size]])] + (db/exec! conn (into [sql] params)))) (defn- clean-deleted - "Walks the pending objects with `db/plan`, a server-side cursor that streams - the rows instead of materialising them, and deletes them chunk by chunk. + "Processes the pending objects chunk by chunk with keyset pagination: each + chunk is one short transaction that reads and locks the next `chunk-size` + objects after the `(deleted_at, id)` cursor and deletes them (database rows + and backend content). - The rows are read in their own read-only transaction and every chunk of - `chunk-size` objects is deleted in its own one (see `delete-in-bulk`), so a - chunk that fails is logged and skipped instead of aborting the whole run, and - the work of the previous chunks is already committed." - [cfg] - (let [xform (comp (map pending-row) - (partition-all chunk-size) - (mapcat group-by-backend)) - rf (fn - ([total] total) - ([total [backend-id ids]] - (+ total (or (delete-in-bulk cfg backend-id ids) 0))))] - (db/tx-run! (assoc cfg ::db/read-only true) - (fn [{:keys [::db/conn]}] - (transduce xform rf 0 - (db/plan conn [sql:get-deleted-sobjects (ct/now)])))))) + A chunk that fails aborts its transaction: nothing of the chunk is deleted, + its locks are released, and it is logged and skipped. The cursor advances + to the `(deleted_at, id)` of the last row of each read chunk regardless of + the deletion result, so a failed chunk is not visited again in the same + run. There is no long-lived read transaction: each transaction lives only + as long as one chunk, letting the vacuum proceed between chunks. Rows + locked by a concurrent run are skipped and deferred to the next run; a + chunk that fails after a partial backend delete self-heals on the next + run (physical deletes are idempotent). Every `throttle-every` deleted + objects the run pauses `throttle-ms` (see `throttle`)." + [{:keys [::mtx/metrics] :as cfg}] + (loop [total 0 + last-deleted-at nil + last-id nil] + + (let [[rows counts] + (db/tx-run! cfg + (fn [{:keys [::db/conn] :as tx}] + (let [rows (fetch-and-lock-chunk conn last-deleted-at last-id)] + (try + [rows (process-chunk-deletion tx rows)] + (catch Throwable cause + ;; Explicit rollback, then return normally (do + ;; not rethrow): the caller needs `rows` to + ;; advance the keyset cursor past the failed + ;; chunk. The commit that `tx-run!` performs + ;; afterwards is a JDBC no-op on the already + ;; rolled-back connection. + (db/rollback! conn) + (l/err :hint "unexpected error on chunk deletion" + :ids (map :id rows) + :cause cause) + [rows {}])))))] + + ;; Reported only after the chunk commits: a rolled-back chunk leaves + ;; an empty count map, so retried rows are never counted twice. + (doseq [[backend-id n] counts] + (mtx/run! metrics + :id :storage-gc-deleted-objects + :inc n + :labels [(name backend-id)])) + + (if-let [cursor (peek rows)] + (let [new-total (+ total (reduce + 0 (vals counts)))] + (throttle total new-total) + (recur new-total + (:deleted-at cursor) + (:id cursor))) + total)))) (defmethod ig/assert-key ::handler [_ params] (assert (sto/valid-storage? (::sto/storage params)) "expect valid storage") - (assert (db/pool? (::db/pool params)) "expect valid storage")) + (assert (db/pool? (::db/pool params)) "expect valid storage") + (assert (mtx/metrics? (::mtx/metrics params)) "expect valid metrics instance")) (defmethod ig/init-key ::handler [_ cfg] diff --git a/backend/test/backend_tests/storage_gc_deleted_test.clj b/backend/test/backend_tests/storage_gc_deleted_test.clj index 9e1f87f723..ecd805af73 100644 --- a/backend/test/backend_tests/storage_gc_deleted_test.clj +++ b/backend/test/backend_tests/storage_gc_deleted_test.clj @@ -10,6 +10,7 @@ [app.common.uuid :as uuid] [app.db :as db] [app.storage :as sto] + [app.storage.impl :as impl] [backend-tests.helpers :as th] [clojure.test :as t])) @@ -91,3 +92,66 @@ (t/testing "the object of the next chunk is gone" (t/is (nil? (find-object last-id)))))) + +(t/deftest failed-batch-is-not-visited-again-in-the-same-run + (let [profile (th/create-profile* 1) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile)}) + + ;; A full first batch of deletable objects. + batch1 (mapv (fn [_] (put-object! (ct/in-past "30m"))) (range chunk-size)) + + ;; The blocked object leads a second full batch, in `deleted_at` order + ;; after the first one: the whole batch fails and must be skipped, and + ;; the cursor must not come back to any of its objects. + blocked (blocked-object! (:id file)) + fillers (mapv (fn [_] (put-object! (ct/in-past "20m"))) (range (dec chunk-size))) + + ;; Objects after the failed batch: the next read batch starts right + ;; after the `(deleted_at, id)` of the last row of the failed one. + batch3 (mapv (fn [_] (put-object! (ct/in-past "5m"))) (range 5)) + + result (th/run-task! :storage-gc-deleted {})] + + (t/testing "the first and the last batches are fully deleted" + (t/is (= (+ (count batch1) (count batch3)) (:deleted result)))) + + (t/testing "the objects of the first batch are gone" + (t/is (every? #(nil? (find-object (:id %))) batch1))) + + (t/testing "the objects of the failed batch are still pending" + (t/is (some? (find-object (:id blocked)))) + (t/is (every? #(some? (find-object (:id %))) fillers))) + + (t/testing "the objects after the failed batch are gone" + (t/is (every? #(nil? (find-object (:id %))) batch3))))) + +(t/deftest physical-failure-rolls-back-the-chunk-and-continues + "Pins the chunk rollback semantics: the physical delete of the first + chunk fails (mocked) after its rows were already deleted inside the + transaction, so the transaction must roll back (the rows stay pending) + while the run skips the failed chunk and deletes the rest." + (let [;; A full first chunk of deletable objects, older in + ;; `deleted_at` order than the rest, so the failing physical + ;; delete hits exactly this chunk. + old (mapv (fn [_] (put-object! (ct/in-past "30m"))) (range chunk-size)) + recent (mapv (fn [_] (put-object! (ct/in-past "5m"))) (range 5)) + + failing (atom true) + orig impl/del-objects-in-bulk + result (with-redefs [impl/del-objects-in-bulk + (fn [backend ids] + (if @failing + (do (reset! failing false) + (throw (ex-info "simulated backend failure" {}))) + (orig backend ids)))] + (th/run-task! :storage-gc-deleted {}))] + + (t/testing "the run finishes and only the objects after the failed chunk are deleted" + (t/is (= (count recent) (:deleted result)))) + + (t/testing "the objects of the failed chunk are still pending" + (t/is (every? #(some? (find-object (:id %))) old))) + + (t/testing "the objects after the failed chunk are gone" + (t/is (every? #(nil? (find-object (:id %))) recent)))))