🐛 Keep the storage GC running past a blocked chunk

`storage-gc-deleted` ran its whole batch inside one transaction: the
handler opened it and `delete-in-bulk` reused that connection, so the
"separate transaction" its comment promised did not exist. A single
row that a foreign key still references aborted that transaction, the
cursor then died on its next fetch ("current transaction is aborted")
and the task ended up throwing `unhandled exception on running task`
after rolling back the whole night of work. That is what has kept
production from deleting a single object since late April.

Now `clean-deleted` walks the pending objects with the cursor in its
own read-only transaction and every chunk runs in its own one, so a
blocked chunk is logged with its ids and skipped while the rest is
committed. `task finished` is reachable again.

The private functions also drop the `!` suffix: the namespace is all
mutation, so the mark adds nothing and reads as noise.

The test builds a blocked object (a `file_media_object` pointing at it
as its thumbnail) plus a full chunk of pending objects and one more in
the next chunk, and checks that the run finishes and that the object of
the next chunk is deleted.

AI-assisted-by: deepseek-v4.1-flash

.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

.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.

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})))

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))))))