diff --git a/.serena/memories/backend/core.md b/.serena/memories/backend/core.md index 816a322a3b..ea34c6beda 100644 --- a/.serena/memories/backend/core.md +++ b/.serena/memories/backend/core.md @@ -87,7 +87,7 @@ Fixtures can populate local data for manual testing/perf work. From the backend * **Type Hinting:** Use explicit JVM type hints (e.g. `^String`, `^long`) in performance-critical paths to avoid reflection overhead. * **Batch inserts:** Use `db/insert-many!` for bulk row inserts — generates a single SQL with multiple parameter tuples. Avoid on very large datasets (SQL length / parameter count limits). -* **Server-side cursors:** Use `db/plan` (fetch-size 1000, forward-only, read-only) or `db/cursor` for large result sets. Never fetch large collections into memory at once. +* **Server-side cursors:** Use `db/plan` (fetch-size 1000, forward-only, read-only) for large result sets. Never fetch large collections into memory at once. `db/plan` is an `IReduceInit`: `reduce`/`transduce` over it (`sequence` does not work on it), and with `partition-all` the transducer's completing arity is what flushes the last partial chunk. A plan row is a **view over the open result set**, valid only inside the reducing function: read the columns you need right there — buffering rows to read them later (a `partition-all` buffer, a lazy seq) silently yields nil columns. `db/cursor` (manual `DECLARE`/`FETCH`) is a leftover with no callers left. * **Transaction discipline:** Use `tx-run!` for writes (opens a transaction), `run!` for reads (single connection, no transaction). Set `:read-only` on `tx-run!` when applicable to let PostgreSQL optimize. ## Lint and Format diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index b0d81330f6..398476a876 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. +- `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-touched` finds references before it sets `deleted_at`. - `objects-gc` removes deleted domain rows and touches their storage object IDs. - Use `::db/reuse-conn true` with `sto/resolve` inside a database transaction. diff --git a/backend/src/app/storage/gc_deleted.clj b/backend/src/app/storage/gc_deleted.clj index c380293453..8356e35b0e 100644 --- a/backend/src/app/storage/gc_deleted.clj +++ b/backend/src/app/storage/gc_deleted.clj @@ -41,22 +41,22 @@ "DELETE FROM storage_object WHERE id = ANY(?::uuid[])") -(defn- delete-sobjects! +(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! +(defn- delete-in-bulk [cfg backend-id ids] - ;; We run the deletion on a separate transaction. This is - ;; because if some exception is raised inside procesing - ;; one chunk, it does not affects the rest of the chunks. + ;; 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)] + (let [total (delete-sobjects conn ids)] (-> (impl/resolve-backend storage backend-id) (impl/del-objects-in-bulk ids)) @@ -82,21 +82,40 @@ AND s.deleted_at <= ? ORDER BY s.deleted_at ASC") -(defn- get-buckets - [conn] - (let [now (ct/now)] - (sequence - (comp (partition-all 25) - (mapcat group-by-backend)) - (db/cursor conn [sql:get-deleted-sobjects now])))) +(def ^:private chunk-size + "Pending objects deleted in each transaction." + 25) -(defn- clean-deleted! - [{:keys [::db/conn] :as cfg}] - (reduce (fn [total [backend-id ids]] - (let [deleted (delete-in-bulk! cfg backend-id ids)] - (+ total (or deleted 0)))) - 0 - (get-buckets conn))) +(defn- pending-row + "Takes the columns the GC needs from a `db/plan` row. + + 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)}) + +(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. + + 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)])))))) (defmethod ig/assert-key ::handler [_ params] @@ -106,7 +125,6 @@ (defmethod ig/init-key ::handler [_ cfg] (fn [_] - (db/tx-run! cfg (fn [cfg] - (let [total (clean-deleted! cfg)] - (l/inf :hint "task finished" :total total) - {:deleted total}))))) + (let [total (clean-deleted cfg)] + (l/inf :hint "task finished" :total total) + {:deleted total}))) diff --git a/backend/test/backend_tests/storage_gc_deleted_test.clj b/backend/test/backend_tests/storage_gc_deleted_test.clj new file mode 100644 index 0000000000..9e1f87f723 --- /dev/null +++ b/backend/test/backend_tests/storage_gc_deleted_test.clj @@ -0,0 +1,93 @@ +;; 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 INC Sucursal en España SL + +(ns backend-tests.storage-gc-deleted-test + (:require + [app.common.time :as ct] + [app.common.uuid :as uuid] + [app.db :as db] + [app.storage :as sto] + [backend-tests.helpers :as th] + [clojure.test :as t])) + +(t/use-fixtures :once th/state-init) +(t/use-fixtures :each th/database-reset) + +;; The GC processes the pending objects in chunks of 25 (see `clean-deleted`), so +;; a chunk boundary is what separates a blocked object from the objects of the +;; next chunk. + +(def ^:private chunk-size 25) + +(defn- put-object! + "Puts a tiny object in storage and returns it." + [deleted-at] + (let [sobject (sto/put-object! (::sto/storage th/*system*) + {::sto/content (sto/content "sobject") + :content-type "text/plain" + :bucket "file-media-object"})] + (th/db-update! :storage-object {:deleted-at deleted-at} {:id (:id sobject)}) + sobject)) + +(defn- blocked-object! + "Puts an object that cannot be deleted because a `file_media_object` points + at it as its thumbnail (a `NO ACTION` foreign key)." + [file-id] + (let [media (sto/put-object! (::sto/storage th/*system*) + {::sto/content (sto/content "media") + :content-type "text/plain" + :bucket "file-media-object"}) + blocked (put-object! (ct/in-past "30m"))] + + (th/db-insert! :file-media-object + {:id (uuid/next) + :file-id file-id + :is-local true + :name "sample" + :media-id (:id media) + :thumbnail-id (:id blocked) + :width 100 + :height 100 + :mtype "image/svg+xml"}) + + blocked)) + +(defn- find-object + [id] + (th/db-get :storage-object {:id id} {::db/remove-deleted false})) + +(t/deftest deletes-the-pending-objects + (let [objects (mapv (fn [_] (put-object! (ct/in-past "10m"))) (range 3)) + result (th/run-task! :storage-gc-deleted {})] + + (t/is (= 3 (:deleted result))) + + (doseq [sobject objects] + (t/is (nil? (find-object (:id sobject))))))) + +(t/deftest blocked-chunk-does-not-stop-the-run + (let [profile (th/create-profile* 1) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile)}) + + ;; The blocked object goes first in `deleted_at` order, so it lands in + ;; the first chunk, together with the objects that fill it up. The last + ;; object of the batch lands in the next chunk: it is the one that + ;; proves the run kept going after the failure. + blocked (blocked-object! (:id file)) + _ (mapv (fn [_] (put-object! (ct/in-past "20m"))) (range (dec chunk-size))) + last-id (:id (put-object! (ct/in-past "5m"))) + + result (th/run-task! :storage-gc-deleted {})] + + (t/testing "the run finishes and only the object of the second chunk is deleted" + (t/is (= 1 (:deleted result)))) + + (t/testing "the blocked object is still pending" + (t/is (some? (find-object (:id blocked))))) + + (t/testing "the object of the next chunk is gone" + (t/is (nil? (find-object last-id))))))