From d0f455d5218bb44ec9e0666796bdbf46c5d299e2 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 16 Oct 2026 10:31:10 +0000 Subject: [PATCH] :zap: Chunk storage-gc-deleted reads with a keyset cursor The storage-gc-deleted task used to stream all the pending rows with db/plan inside one long read-only transaction, blocking the table vacuum for the whole run and generating bloat on long deletions. Now every chunk of 25 objects is one short read-write transaction: the SELECT reads and locks the next objects after a (deleted_at, id) keyset cursor with FOR UPDATE SKIP LOCKED, deletes the database rows grouped by backend and removes their backend content. Concurrent runs never take the same chunk, and the double SELECT of the old lock-ids step is gone. A chunk that fails aborts its transaction: nothing of the chunk is deleted, its locks are released, and the run skips it; the cursor still advances to the last row read, so a failed chunk is not visited again in the same run. AI-assisted-by: glm-5.3-flash --- .serena/memories/backend/storage.md | 2 +- backend/src/app/main.clj | 9 +- backend/src/app/storage/gc_deleted.clj | 208 +++++++++++------- .../backend_tests/storage_gc_deleted_test.clj | 64 ++++++ 4 files changed, 204 insertions(+), 79 deletions(-) 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)))))