From 1783f9380d07a0dd6ae980cad9b979a65d39e69f Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Thu, 15 Oct 2026 14:18:57 +0200 Subject: [PATCH] :sparkles: 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 | 5 +- backend/src/app/storage/gc_deleted.clj | 30 +--- backend/src/app/storage/gc_touched.clj | 53 +++++- backend/test/backend_tests/rpc_font_test.clj | 6 +- backend/test/backend_tests/storage_test.clj | 169 ++++++++++++++++++- 5 files changed, 227 insertions(+), 36 deletions(-) diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index 3d978642df..816e254b2b 100644 --- a/.serena/memories/backend/storage.md +++ b/.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. | diff --git a/backend/src/app/storage/gc_deleted.clj b/backend/src/app/storage/gc_deleted.clj index 6c7e419676..22677ab429 100644 --- a/backend/src/app/storage/gc_deleted.clj +++ b/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" diff --git a/backend/src/app/storage/gc_touched.clj b/backend/src/app/storage/gc_touched.clj index f73b6024f1..104f9b7f17 100644 --- a/backend/src/app/storage/gc_touched.clj +++ b/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 diff --git a/backend/test/backend_tests/rpc_font_test.clj b/backend/test/backend_tests/rpc_font_test.clj index 90dabf62a0..bd81112bcb 100644 --- a/backend/test/backend_tests/rpc_font_test.clj +++ b/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)} diff --git a/backend/test/backend_tests/storage_test.clj b/backend/test/backend_tests/storage_test.clj index abdd8c1503..f85319bc47 100644 --- a/backend/test/backend_tests/storage_test.clj +++ b/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.