✨ Scan upload-session chunks in storage gc-touched (#12305)

Route upload-session chunk liveness through the mark-time reference
scan instead of the delete-time cleanup in gc-deleted.

gc-touched now checks upload_session_chunk before marking a chunk,
so referenced chunks freeze and only orphans are marked. The freeze
path re-verifies references under row locks, which also closes the
stale-freeze race for every scanned bucket. gc-deleted drops its
upload-session special case: deferred rows keep their mappings until
the final give-up, and a live reference aborts the chunk loud.

Closes #12302

AI-assisted-by: muse-spark-1.3-contributor

.serena/memories/backend/storage.md

@@ -25,8 +25,9 @@
 - Existing object values can remain readable until physical deletion.
 - `:expired-at` blocks reads after the expiration time.
 - `del-object!` sets `deleted_at` on live rows only (`deleted_at IS NULL`): a repeated call returns `false`. 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. A physical-deletion failure inside a chunk defers its rows one day (`deletion_attempts + 1`) and gives up after 7 attempts, accepting the orphan blob; chunk mappings (`upload_session_chunk`) are removed before the `storage_object` rows (NO ACTION foreign keys).
+- `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. A physical-deletion failure inside a chunk defers its rows one day (`deletion_attempts + 1`) and gives up after 7 attempts, accepting the orphan blob. Deferred rows keep their references until the final give-up. The freeze path re-verifies references under row locks, so a concurrent purge can delay a freeze but never swallow a touch.
 - `storage-gc-touched` finds references before it sets `deleted_at`.
+- Orphan watch: `SELECT COALESCE(metadata->>'bucket', metadata->>'~:bucket') AS bucket, count(*) FROM storage_object so WHERE touched_at IS NULL AND deleted_at IS NULL AND status = 'valid' AND NOT EXISTS (SELECT 1 FROM upload_session_chunk usc WHERE usc.object_id = so.id) GROUP BY 1;` Only the `upload-session` row of that output is a leak signal (its rows always have a mapping or a pending touch/deleted mark); other buckets use domain-table references, not this mapping.
 - `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.
 
@@ -83,7 +84,7 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
 | `profile` | User and team profile photos. References: `profile.photo_id` and `team.photo_id`. | Yes | Authentication required | Reference scan. |
 | `organization` | Organization logos uploaded by the Nitrate management API. | Yes | Public | No reference scan. A touched object is deleted. |
 | `tempfile` | Export files and temporary font downloads. | No | Authentication required | No reference scan. A touched object uses a two-hour deletion delay. |
-| `upload-session` | Chunked-upload chunks. References: `upload_session_chunk.object_id` and `upload_session_chunk.session_id` (both NO ACTION DEFERRABLE: restrict semantics, procedural deletion). | No | Authentication required | No reference scan. A touched object is deleted after the delay; `gc-deleted` removes mappings before rows. |
+| `upload-session` | Chunked-upload chunks. References: `upload_session_chunk.object_id` and `upload_session_chunk.session_id` (both NO ACTION DEFERRABLE: restrict semantics, procedural deletion). | No | Authentication required | Reference scan on `upload_session_chunk.object_id`; a referenced chunk freezes, an orphan is deleted after the delay. Deferred rows keep their mappings until give-up. |
 | `file-data` | Encoded file data when `file-data-backend` is `storage`. Reference metadata has `storage-ref-id`, `file-id`, and the `file_data` row ID. | Yes | Authentication required | Reference scan. |
 | `file-data-fragment` | Compatibility value for file-data fragments. The current backend has no dedicated producer for this bucket. | No current write semantics | Public | No touched-object collector case. |
 | `file-change` | Compatibility value for file changes. Current snapshots store data in `file_data`, not this bucket. | No current write semantics | Authentication required | No touched-object collector case. |

backend/src/app/storage/gc_deleted.clj

@@ -57,18 +57,6 @@
     (-> (db/exec-one! conn [sql:delete-sobjects ids])
         (db/get-update-count))))
 
-(def ^:private sql:delete-upload-session-chunks
-  "DELETE FROM upload_session_chunk
-    WHERE object_id = ANY(?::uuid[])")
-
-(defn- delete-upload-session-chunks
-  "Remove the chunk mappings for the given storage object ids. This must run
-  before the storage_object rows are deleted: the upload_session_chunk
-  foreign keys are ON DELETE NO ACTION."
-  [conn ids]
-  (let [ids (db/create-array conn "uuid" ids)]
-    (db/exec-one! conn [sql:delete-upload-session-chunks ids])))
-
 (def ^:private sql:increment-attempts-and-defer
   "UPDATE storage_object
       SET deletion_attempts = deletion_attempts + 1,
@@ -117,21 +105,17 @@
                :backend (name backend-id)))
 
       (when (seq ok-ids)
-        ;; NOTE: the chunk mappings must be removed before the
-        ;; storage_object rows (NO ACTION foreign keys). It only affects
-        ;; objects of the upload-session bucket; for any other bucket the
-        ;; delete matches no rows.
-        (delete-upload-session-chunks conn ok-ids)
+        ;; NOTE: rows arrive here with no live references (the mark-time
+        ;; scan in gc-touched owns that check, also for the upload-session
+        ;; bucket). If a NO ACTION reference still exists, this DELETE
+        ;; aborts the chunk loud instead of breaking anyone silently.
         (delete-sobjects conn ok-ids))
 
       (when (seq fail-ids)
+        ;; NOTE: deferred rows keep both their row and their references
+        ;; (upload-session mappings included) until the final give-up, so
+        ;; a backend outage does not break live sessions.
         (increment-attempts-and-defer conn fail-ids)
-        ;; NOTE: same NO ACTION ordering as above: the give-up DELETE below
-        ;; removes storage_object rows, so chunk mappings must go first.
-        ;; Deferred objects keep their rows; only the mapping of a
-        ;; permanently given-up object disappears early, and that object is
-        ;; already deleted-marked.
-        (delete-upload-session-chunks conn fail-ids)
         (let [given-up (delete-give-up conn fail-ids)]
           (when (pos? (db/get-update-count given-up))
             (l/wrn :hint "giving up on orphan blob after max attempts"

backend/src/app/storage/gc_touched.clj

@@ -77,6 +77,15 @@
   (-> (db/exec-one! conn [sql:has-file-thumbnail-refs id])
       (get :has-refs)))
 
+(def ^:private
+  sql:has-upload-session-chunk-refs
+  "SELECT EXISTS (SELECT 1 FROM upload_session_chunk WHERE object_id = ?) AS has_refs")
+
+(defn- has-upload-session-chunk-refs?
+  [conn {:keys [id]}]
+  (-> (db/exec-one! conn [sql:has-upload-session-chunk-refs id])
+      (get :has-refs)))
+
 (def sql:exists-file-data-refs
   "SELECT EXISTS (
      SELECT 1 FROM file_data
@@ -146,6 +155,31 @@
                :id (str id))
         sto/default-bucket)))
 
+(def ^:private sql:lock-storage-objects
+  "SELECT id FROM storage_object WHERE id = ANY(?::uuid[]) ORDER BY id FOR UPDATE")
+
+(defn- verify-frozen
+  "Re-checks, holding the row locks, that the about-to-freeze objects are
+  still referenced. A concurrent purge (mapping delete plus touch) either
+  commits before the check and the object moves to deletion, or blocks on
+  the lock and its touch wins last, so no touch is ever swallowed by a
+  stale freeze. Ids are locked in order to avoid lock-order inversions
+  between workers. Returns [still-frozen-ids now-deletable-ids]."
+  [conn has-refs? bucket objects]
+  (let [by-id  (into {} (map (juxt :id identity)) objects)
+        ids    (db/create-array conn "uuid" (sort (keys by-id)))
+        locked (into #{} (map :id) (db/exec! conn [sql:lock-storage-objects ids]))]
+    (reduce (fn [[frozen deletable] {:keys [id] :as object}]
+              (if (and (contains? locked id) (has-refs? conn object))
+                [(conj frozen id) deletable]
+                (do
+                  (l/dbg :id (str id)
+                         :status "delete-after-recheck"
+                         :bucket bucket)
+                  [frozen (conj deletable id)])))
+            [#{} #{}]
+            objects)))
+
 (defn- process-objects
   [conn has-refs? bucket objects]
   (loop [to-freeze #{}
@@ -157,17 +191,24 @@
           (l/dbg :id (str id)
                  :status "freeze"
                  :bucket bucket)
-          (recur (conj to-freeze id) to-delete (rest objects)))
+          (recur (conj to-freeze object) to-delete (rest objects)))
         (do
           (l/dbg :id (str id)
                  :status "delete"
                  :bucket bucket)
           (recur to-freeze (conj to-delete id) (rest objects))))
-      (do
-        (some->> (seq to-freeze) (mark-freeze-in-bulk conn))
-        (some->> (seq to-delete) (mark-delete-in-bulk conn))
-        [(count to-freeze) (count to-delete)]))))
+      (let [[frozen deletable] (if (seq to-freeze)
+                                 (verify-frozen conn has-refs? bucket to-freeze)
+                                 [#{} #{}])
+            doomed (into to-delete deletable)]
+        (some->> (seq frozen) (mark-freeze-in-bulk conn))
+        (some->> (seq doomed) (mark-delete-in-bulk conn))
+        [(count frozen) (count doomed)]))))
 
+;; NOTE: upload-session chunks are 1:1 with their mapping and never
+;; shared, but the scan owns the liveness decision anyway: a referenced
+;; chunk freezes, so gc-deleted never meets a live NO ACTION reference.
+;; The freeze path re-verifies under row locks (see verify-frozen).
 (defn- process-bucket
   [conn bucket objects]
   (cond
@@ -178,7 +219,7 @@
     (= bucket "profile")                 (process-objects conn has-profile-refs? bucket objects)
     (= bucket "file-data")               (process-objects conn has-file-data-refs? bucket objects)
     (= bucket sto/tempfile-bucket)       (process-objects conn (constantly false) sto/tempfile-bucket objects)
-    (= bucket sto/upload-session-bucket) (process-objects conn (constantly false) sto/upload-session-bucket objects)
+    (= bucket sto/upload-session-bucket) (process-objects conn has-upload-session-chunk-refs? sto/upload-session-bucket objects)
     (= bucket "organization")            (process-objects conn (constantly false) bucket objects)
     :else
     (ex/raise :type :internal

backend/test/backend_tests/rpc_font_test.clj

@@ -140,7 +140,7 @@
 
     (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
                 (th/run-task! :storage-gc-touched {}))]
-      (t/is (= 6 (:freeze res))))
+      (t/is (= 8 (:freeze res))))
 
     (let [params {::th/type :delete-font
                   ::rpc/profile-id (:id prof)
@@ -209,7 +209,7 @@
 
     (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
                 (th/run-task! :storage-gc-touched {}))]
-      (t/is (= 6 (:freeze res))))
+      (t/is (= 8 (:freeze res))))
 
     (let [params {::th/type :delete-font
                   ::rpc/profile-id (:id prof)
@@ -260,7 +260,7 @@
     ;; freeze with hours 3 clock
     (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
                 (th/run-task! :storage-gc-touched {}))]
-      (t/is (= 6 (:freeze res))))
+      (t/is (= 8 (:freeze res))))
 
     (let [params {::th/type :delete-font-variant ::rpc/profile-id (:id prof)
                   :team-id team-id :id (-> out1 :result :id)}

backend/test/backend_tests/storage_test.clj

@@ -16,6 +16,7 @@
    [app.rpc :as-alias rpc]
    [app.storage :as sto]
    [app.storage.fs :as-alias sto.fs]
+   [app.storage.gc-touched :as gc-touched]
    [app.storage.impl :as impl]
    [app.storage.s3 :as-alias sto.s3]
    [app.storage.schema :as stsch]
@@ -328,8 +329,10 @@
     ;; run the touched gc task
     (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
                 (th/run-task! :storage-gc-touched {}))]
-      (t/is (= 5 (:freeze res)))
-      (t/is (= 1 (:delete res)))
+      (t/is (= 6 (:freeze res)))
+      ;; the consumed-but-not-yet-purged session keeps its mapping,
+      ;; so its chunk freezes instead of being marked
+      (t/is (= 0 (:delete res)))
 
       (let [result-1 (:result out1)
             result-2 (:result out2)]
@@ -1108,6 +1111,168 @@
       (t/is (= 1 (:deleted res))))
     (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object"]))))))
 
+(t/deftest upload-session-chunk-frozen-while-referenced
+  ;; A chunk whose session mapping is still alive must be frozen, not
+  ;; marked for deletion: the reference scan owns the liveness decision.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        _          (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})]
+    (t/is (nil? (:error out)))
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+
+    ;; the session is neither consumed nor stalled: the mapping is alive
+    (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
+                (th/run-task! :storage-gc-touched {}))]
+      (t/is (= 1 (:freeze res)))
+      (t/is (= 0 (:delete res))))
+
+    ;; frozen: untouched, not marked, mapping intact
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object where deleted_at is not null"]))))
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))))
+
+(t/deftest upload-session-freeze-reverify
+  ;; The freeze recheck runs under row locks: an object whose mapping
+  ;; disappeared (plus a concurrent touch) after the first check is
+  ;; reclassified to deletion instead of swallowing the touch.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        _          (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})
+        object-id  (:object-id (th/db-exec-one! ["select object_id from upload_session_chunk where session_id = ?"
+                                                 session-id]))
+        check      @#'gc-touched/has-upload-session-chunk-refs?
+        verify     @#'gc-touched/verify-frozen]
+    (t/is (nil? (:error out)))
+
+    ;; mapping alive: stays frozen
+    (let [[frozen deletable] (db/transact! th/*pool* (fn [conn] (verify conn check "upload-session" [{:id object-id}])))]
+      (t/is (= #{object-id} frozen))
+      (t/is (= #{} deletable)))
+
+    ;; simulate a purge that committed after the first check: mapping
+    ;; gone, touch bumped
+    (th/db-exec! ["delete from upload_session_chunk where session_id = ?" session-id])
+    (th/db-exec! ["update storage_object set touched_at = ? where id = ?" (ct/now) object-id])
+
+    ;; recheck moves it to deletion
+    (let [[frozen deletable] (db/transact! th/*pool* (fn [conn] (verify conn check "upload-session" [{:id object-id}])))]
+      (t/is (= #{} frozen))
+      (t/is (= #{object-id} deletable)))))
+
+(t/deftest upload-session-gc-deleted-skips-live-mapping
+  ;; A row marked for deletion while its mapping is still alive must NOT
+  ;; be reaped silently: the chunk is skipped loud (logged) and retried
+  ;; once objects-gc purges the session. This pins the mark-time
+  ;; invariant instead of breaking sessions quietly.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        _          (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})
+        object-id  (:object-id (th/db-exec-one! ["select object_id from upload_session_chunk where session_id = ?"
+                                                 session-id]))]
+    (t/is (nil? (:error out)))
+
+    ;; force the anomaly: mark deleted while the mapping is alive
+    (th/db-exec! ["update storage_object set deleted_at = ? where id = ?" (ct/now) object-id])
+
+    (let [res (th/run-task! :storage-gc-deleted {})]
+      (t/is (= 0 (:deleted res))))
+
+    ;; row and mapping both intact for the retry
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where id = ?" object-id]))))
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))))
+
+(t/deftest upload-session-deferred-keeps-mapping
+  ;; A physical-deletion failure defers the row; the session mapping is
+  ;; kept so the session survives the backend outage and the blob is
+  ;; retried the next day.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        _          (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})
+        object-id  (:object-id (th/db-exec-one! ["select object_id from upload_session_chunk where session_id = ?"
+                                                 session-id]))]
+    (t/is (nil? (:error out)))
+    (th/db-exec! ["update storage_object set deleted_at = ? where id = ?" (ct/now) object-id])
+
+    (with-mocks [_mock {:target 'app.storage.impl/del-objects-in-bulk
+                        :return (fn [_ ids] (set ids))}]
+      (let [res (th/run-task! :storage-gc-deleted {})]
+        (t/is (= 0 (:deleted res)))))
+
+    ;; row kept and deferred one day
+    (let [row (th/db-exec-one! ["select deletion_attempts, touched_at from storage_object where id = ?" object-id])]
+      (t/is (= 1 (:deletion-attempts row)))
+      (t/is (some? (:touched-at row))))
+    ;; mapping kept: the session is intact for the retry
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))))
+
 (t/deftest upload-session-consumed-purge
   ;; An assembled session is marked as consumed and objects-gc purges it
   ;; right away, without waiting for the stalled threshold.