diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index b0d81330f6..101c77702a 100644 --- a/.serena/memories/backend/storage.md +++ b/.serena/memories/backend/storage.md @@ -30,10 +30,46 @@ - `objects-gc` removes deleted domain rows and touches their storage object IDs. - Use `::db/reuse-conn true` with `sto/resolve` inside a database transaction. +## Connection Reuse Details + +### `app.storage/resolve` patterns: + +**1. Pool mode (default)** - `(sto/resolve cfg)` +- Returns storage abstraction from config +- Uses whatever database pool is available +- **Safe to call outside transaction context** +- Used in: `rpc/commands/media.clj:363`, `rpc/commands/auth.clj:327`, `rpc/commands/profile.clj:362` + +**2. Connection reuse mode** - `(sto/resolve cfg ::db/reuse-conn true)` +- Internally calls `db/get-connection cfg` to obtain connectable +- Configures storage with the specific connection from config +- **Must be paired with transaction that owns this connection** +- Used in: `features/fdata.clj:100`, `rpc/commands/media.clj:425`, `rpc/commands/files_thumbnails.clj:307,319`, `binfile/v3.clj:722` + +**3. Explicit configuration** - `(sto/configure storage conn)` +- Sets `::db/conn` on storage map directly +- Asserts `db/conn? connection` (storage.clj:349) +- Used inside `db/tx-run!` blocks where `conn` is already available +- Used in: `tasks/file_gc.clj:256`, `rpc/commands/files_thumbnails.clj:347,371` + +### Key Warning (from function notes): + +The improved note in `import-storage-objects` and `handle-persistence` warns: +**Do not reuse the main database connection for storage operations within a transaction.** The storage upload process can fail mid-operation, leaving orphaned objects on the backend. If the outer transaction aborts, pending storage objects become unreconciliable because the storage subsystem registers its pending state in separate transactions. + +### Rule of Thumb for `sto/put-object!`: + +Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `impl/put-object`) and does not directly use `::db/conn` or `::db/pool`, **all usage of `put-object!` will never run inside a common transaction** (if configured at all). The storage backend operations are independent of the database transaction boundary. + ## Deduplication - Deduplication requires `::sto/deduplicate?`, a content hash, and bucket metadata. - The lookup matches hash, bucket, backend, and `deleted_at IS NULL`. +- The lookup only considers rows with `status='valid'`; pending rows are invisible. +- A hit whose blob is missing is repaired in place: the same row/id is kept, + and `put-object!` rewrites the blob under that id. This heals all existing + references to the object. If the rewrite fails, the row is left live and + valid for a later retry. - The lookup does not include file ID, profile ID, team ID, or organization ID. - Objects can therefore share content across users and files within one bucket. - Deleted objects are not reused. diff --git a/backend/src/app/binfile/common.clj b/backend/src/app/binfile/common.clj index 2f37e34e78..16652c3526 100644 --- a/backend/src/app/binfile/common.clj +++ b/backend/src/app/binfile/common.clj @@ -27,7 +27,6 @@ [app.features.file-migrations :as fmigr] [app.loggers.audit :as-alias audit] [app.loggers.webhooks :as-alias webhooks] - [app.storage :as sto] [app.util.blob :as blob] [app.util.pointer-map :as pmap] [app.worker :as-alias wrk] @@ -654,27 +653,6 @@ (db/exec-one! conn ["SET LOCAL idle_in_transaction_session_timeout = 0"]) (db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"]))) -(defn invalidate-thumbnails - [cfg file-id] - (let [storage (sto/resolve cfg) - - sql-1 - (str "update file_tagged_object_thumbnail " - " set deleted_at = now() " - " where file_id=? returning media_id") - - sql-2 - (str "update file_thumbnail " - " set deleted_at = now() " - " where file_id=? returning media_id")] - - (run! #(sto/touch-object! storage %) - (sequence - (keep :media-id) - (concat - (db/exec! cfg [sql-1 file-id]) - (db/exec! cfg [sql-2 file-id])))))) - (defn process-file [cfg {:keys [id] :as file}] (let [libs (delay (get-resolved-file-libraries cfg file))] diff --git a/backend/src/app/binfile/v3.clj b/backend/src/app/binfile/v3.clj index b6f610000d..4fd561041f 100644 --- a/backend/src/app/binfile/v3.clj +++ b/backend/src/app/binfile/v3.clj @@ -866,6 +866,13 @@ [{:keys [::bfc/input ::entries ::bfc/timestamp] :as cfg}] (events/tap :progress {:section :storage-objects}) + ;; IMPORTANT: we strongly do not reuse the main connection that can + ;; run inside a transaction because the storage upload process can + ;; fail in the middle of uploading and leave garbage on the underlying + ;; backend, if we participate in the main transaction and it aborts + ;; we will lose all registry of the pending to reconcile blobs + ;; what the storage subsystem registers in other parallel + ;; transaction (let [storage (sto/resolve cfg) entries (keep (match-storage-entry-fn) entries)] @@ -1051,6 +1058,27 @@ {:file-ids file-ids :resolution resolution}))) +(defn- invalidate-thumbnails + [cfg file-id] + (let [storage (sto/resolve cfg ::db/reuse-conn true) + + sql-1 + (str "update file_tagged_object_thumbnail " + " set deleted_at = now() " + " where file_id=? returning media_id") + + sql-2 + (str "update file_thumbnail " + " set deleted_at = now() " + " where file_id=? returning media_id")] + + (run! #(sto/touch-object! storage %) + (sequence + (keep :media-id) + (concat + (db/exec! cfg [sql-1 file-id]) + (db/exec! cfg [sql-2 file-id])))))) + (defn- import-file-and-overwrite* [{:keys [::manifest ::bfc/file-id] :as cfg}] @@ -1074,7 +1102,7 @@ (import-storage-objects cfg) (import-file cfg file) - (bfc/invalidate-thumbnails cfg file-id) + (invalidate-thumbnails cfg file-id) (bfm/apply-pending-migrations! cfg) {:file-ids [file-id] diff --git a/backend/src/app/features/fdata.clj b/backend/src/app/features/fdata.clj index e65aa5d930..43484a696b 100644 --- a/backend/src/app/features/fdata.clj +++ b/backend/src/app/features/fdata.clj @@ -151,6 +151,13 @@ (cond (= backend "storage") + ;; IMPORTANT: we strongly do not reuse the main connection that can + ;; run inside a transaction because the storage upload process can + ;; fail in the middle of uploading and leave garbage on the underlying + ;; backend, if we participate in the main transaction and it aborts + ;; we will lose all registry of the pending to reconcile blobs + ;; what the storage subsystem registers in other parallel + ;; transaction (let [storage (sto/resolve cfg) content (sto/content data) sobject (sto/put-object! storage diff --git a/backend/src/app/features/file_snapshots.clj b/backend/src/app/features/file_snapshots.clj index 6bc70fd52d..9be4b1b572 100644 --- a/backend/src/app/features/file_snapshots.clj +++ b/backend/src/app/features/file_snapshots.clj @@ -326,8 +326,11 @@ (let [file (d/update-when row :metadata fdata/decode-metadata) vern (rand-int Integer/MAX_VALUE) + ;; We reuse the main connection here for storage operations + ;; becaue the main operations are touching and we need them + ;; to be atomic with the current transaction storage - (sto/resolve cfg {::db/reuse-conn true}) + (sto/resolve cfg ::db/reuse-conn true) snapshot (get-snapshot cfg file-id snapshot-id)] diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index dc51498bb3..f1aadd0a2d 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -37,6 +37,7 @@ [app.storage.fs :as-alias sto.fs] [app.storage.gc-deleted :as-alias sto.gc-deleted] [app.storage.gc-touched :as-alias sto.gc-touched] + [app.storage.pending-gc :as-alias sto.pending-gc] [app.storage.s3 :as-alias sto.s3] [app.system :as sys] [app.util.cron] @@ -199,6 +200,10 @@ ::sto.gc-touched/handler {::db/pool (ig/ref ::db/pool)} + ::sto.pending-gc/handler + {::db/pool (ig/ref ::db/pool) + ::sto/storage (ig/ref ::sto/storage)} + ::http.client/client {} @@ -386,6 +391,7 @@ :upload-session-gc (ig/ref :app.tasks.upload-session-gc/handler) :storage-gc-deleted (ig/ref ::sto.gc-deleted/handler) :storage-gc-touched (ig/ref ::sto.gc-touched/handler) + :storage-pending-gc (ig/ref ::sto.pending-gc/handler) :session-gc (ig/ref ::session.tasks/gc) :audit-log-archive (ig/ref :app.loggers.audit.archive-task/handler) :audit-log-gc (ig/ref :app.loggers.audit.gc-task/handler) @@ -545,6 +551,9 @@ {:cron #penpot/cron "0 0 0 * * ?" ;; daily :task :storage-gc-touched} + {:cron #penpot/cron "0 0 0 * * ?" ;; daily + :task :storage-pending-gc} + {:cron #penpot/cron "0 0 0 * * ?" ;; daily :task :tasks-gc} diff --git a/backend/src/app/migrations.clj b/backend/src/app/migrations.clj index 9255a14f03..ff65057bff 100644 --- a/backend/src/app/migrations.clj +++ b/backend/src/app/migrations.clj @@ -499,7 +499,10 @@ :fn (mg/resource "app/migrations/sql/0152-improve-uuid-defaults-and-drop-extension.sql")} {:name "0152-rename-version-and-add-indexes-to-server-error-report" - :fn (mg/resource "app/migrations/sql/0152-rename-version-and-add-indexes-to-server-error-report.sql")}]) + :fn (mg/resource "app/migrations/sql/0152-rename-version-and-add-indexes-to-server-error-report.sql")} + + {:name "0153-add-storage-object-status-and-deletion-attempts" + :fn (mg/resource "app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql")}]) (defn apply-migrations! [pool name migrations] diff --git a/backend/src/app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql b/backend/src/app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql new file mode 100644 index 0000000000..68efad792a --- /dev/null +++ b/backend/src/app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql @@ -0,0 +1,24 @@ +--- Add a status column and a deletion attempts counter to storage_object. + +--- The status column tracks the write-ahead lifecycle of newly created +--- objects. A row is inserted as 'pending' before its blob is written to +--- the underlying storage subsystem and promoted to 'valid' once the +--- write succeeds. Rows in 'pending' state are excluded from the normal +--- lifecycle (deduplication, gc, reads) until they become valid; a +--- periodic task (:storage-pending-gc) reclaims pending rows that were +--- never promoted (e.g. after a crash). + +ALTER TABLE storage_object + ADD COLUMN status text NOT NULL DEFAULT 'valid' + CHECK (status IN ('valid', 'pending')); + +CREATE INDEX storage_object__status_created_at__idx + ON storage_object (status, created_at) + WHERE status = 'pending'; + +--- The deletion_attempts counter tracks how many times the gc_deleted +--- task has attempted to physically delete the blob. After max attempts +--- the row is removed and the blob is left as an orphan. + +ALTER TABLE storage_object + ADD COLUMN deletion_attempts bigint NOT NULL DEFAULT 0; diff --git a/backend/src/app/rpc/commands/files_thumbnails.clj b/backend/src/app/rpc/commands/files_thumbnails.clj index 74d277ee79..1a821c4027 100644 --- a/backend/src/app/rpc/commands/files_thumbnails.clj +++ b/backend/src/app/rpc/commands/files_thumbnails.clj @@ -299,30 +299,32 @@ ;; --- MUTATION COMMAND: delete-file-object-thumbnail (defn- delete-file-object-thumbnail! - [{:keys [::db/conn ::sto/storage]} file-id object-id] + [{:keys [::db/conn] :as cfg} file-id object-id] (when-let [{:keys [media-id tag]} (db/get* conn :file-tagged-object-thumbnail {:file-id file-id :object-id object-id} {::sql/for-update true})] - (sto/touch-object! storage media-id) - (db/update! conn :file-tagged-object-thumbnail - {:deleted-at (ct/now)} - {:file-id file-id - :object-id object-id - :tag tag}))) + (let [storage (sto/resolve cfg ::db/reuse-conn true)] + (sto/touch-object! storage media-id) + (db/update! conn :file-tagged-object-thumbnail + {:deleted-at (ct/now)} + {:file-id file-id + :object-id object-id + :tag tag})))) (defn- delete-file-object-thumbnails! "Soft-deletes multiple object thumbnails in a single UPDATE statement with RETURNING, then touches all returned media objects." - [{:keys [::db/conn ::sto/storage]} object-ids] - (let [ids (db/create-array conn "text" (seq object-ids)) - sql (str/concat - "UPDATE file_tagged_object_thumbnail" - " SET deleted_at = now()" - " WHERE object_id = ANY(?)" - " AND deleted_at IS NULL" - " RETURNING media_id") - rows (db/exec! conn [sql ids])] + [{:keys [::db/conn] :as cfg} object-ids] + (let [storage (sto/resolve cfg ::db/reuse-conn true) + ids (db/create-array conn "text" (seq object-ids)) + sql (str/concat + "UPDATE file_tagged_object_thumbnail" + " SET deleted_at = now()" + " WHERE object_id = ANY(?)" + " AND deleted_at IS NULL" + " RETURNING media_id") + rows (db/exec! conn [sql ids])] (doseq [{:keys [media-id]} rows] (sto/touch-object! storage media-id)))) @@ -342,10 +344,8 @@ ::audit/skip true} [cfg {:keys [::rpc/profile-id file-id object-id]}] (files/check-edition-permissions! cfg profile-id file-id) - (db/tx-run! cfg (fn [{:keys [::db/conn] :as cfg}] - (-> cfg - (update ::sto/storage sto/configure conn) - (delete-file-object-thumbnail! file-id object-id)) + (db/tx-run! cfg (fn [cfg] + (delete-file-object-thumbnail! cfg file-id object-id) nil))) (sv/defmethod ::delete-file-object-thumbnails @@ -366,11 +366,7 @@ (doseq [file-id file-ids] (files/check-edition-permissions! conn profile-id file-id)))) ;; Delete all matching thumbnails in one transaction - (db/tx-run! cfg (fn [{:keys [::db/conn] :as cfg}] - (-> cfg - (update ::sto/storage sto/configure conn) - (delete-file-object-thumbnails! object-ids)) - nil))))) + (db/tx-run! cfg delete-file-object-thumbnails! object-ids)))) ;; --- MUTATION COMMAND: create-file-thumbnail diff --git a/backend/src/app/rpc/commands/media.clj b/backend/src/app/rpc/commands/media.clj index a7bd2ec7fe..7bf42c57a9 100644 --- a/backend/src/app/rpc/commands/media.clj +++ b/backend/src/app/rpc/commands/media.clj @@ -377,7 +377,7 @@ (sto/put-object! storage {::sto/content data ::sto/deduplicate? false - ::sto/touch true + ::sto/touched-at (ct/in-future {:hours 1}) :content-type (:mtype content) :bucket sto/tempfile-bucket :upload-id (str session-id) @@ -393,6 +393,7 @@ FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND deleted_at IS NULL + AND status = 'valid' ORDER BY (metadata->>'~:chunk-index')::integer ASC") (defn- get-upload-chunks diff --git a/backend/src/app/storage.clj b/backend/src/app/storage.clj index 2ae8bf5632..0f35b0a54a 100644 --- a/backend/src/app/storage.clj +++ b/backend/src/app/storage.clj @@ -70,7 +70,7 @@ [:map {:title "storage"} [::backends schema:backends] [::backend [:enum :s3 :fs]] - ::db/connectable]) + ::db/pool]) (def valid-storage? (sm/validator schema:storage)) @@ -96,7 +96,7 @@ (-> (d/without-nils cfg) (assoc ::backends backends) (assoc ::backend backend) - (assoc ::db/connectable pool)))) + (assoc ::db/pool pool)))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; Database Objects @@ -118,60 +118,26 @@ " and (metadata->>'~:bucket') = ? " " and backend = ?" " and deleted_at is null" + " and status = 'valid'" " limit 1")] - (some-> (db/exec-one! connectable [sql hash bucket (name backend)]) - (update :metadata db/decode-transit-pgobject)))) + ;; NOTE: metadata is left encoded; row->storage-object is + ;; responsible for decoding it. + (db/exec-one! connectable [sql hash bucket (name backend)]))) -(defn- create-database-object - [{:keys [::backend ::db/connectable]} {:keys [::content ::expired-at ::touched-at ::touch] :as params}] - (let [id (or (::id params) (uuid/random)) - mdata (cond-> (get-metadata params) - (satisfies? impl/IContentHash content) - (assoc :hash (impl/get-hash content))) - - touched-at (if touch - (or touched-at (ct/now)) - touched-at) - - ;; NOTE: for now we don't reuse the deleted objects, but in - ;; futute we can consider reusing deleted objects if we - ;; found a duplicated one and is marked for deletion but - ;; still not deleted. - result (when (and (::deduplicate? params) - (:hash mdata) - (:bucket mdata) - (not= tempfile-bucket (:bucket mdata))) - (let [result (get-database-object-by-hash connectable backend - (:bucket mdata) - (:hash mdata))] - (if touch - (do - (db/update! connectable :storage-object - {:touched-at touched-at} - {:id (:id result)} - {::db/return-keys false}) - (assoc result :touced-at touched-at)) - result))) - - result (or result - (-> (db/insert! connectable :storage-object - {:id id - :size (impl/get-size content) - :backend (name backend) - :metadata (db/tjson mdata) - :deleted-at expired-at - :touched-at touched-at}) - (update :metadata db/decode-transit-pgobject) - (update :metadata assoc ::created? true)))] - - (impl/storage-object - (:id result) - (:size result) - (:created-at result) - (:deleted-at result) - (:touched-at result) - backend - (:metadata result)))) +(defn- promote-object! + [storage object] + (let [ds (db/get-connectable storage) + res (-> (db/update! ds :storage-object + {:status "valid"} + {:id (:id object)} + {::db/return-keys false}) + (db/get-update-count))] + (when-not (pos? res) + ;; The pending row disappeared while the blob was being written + ;; (e.g. reclaimed by :storage-pending-gc); make it observable. + (l/wrn :hint "unable to promote storage object, pending row not found" + :id (str (:id object)))) + res)) (defn row->storage-object [res] (let [mdata (or (some-> (:metadata res) (db/decode-transit-pgobject)) {})] @@ -188,7 +154,8 @@ "SELECT * FROM storage_object WHERE id = ? - AND (deleted_at IS NULL)") + AND (deleted_at IS NULL) + AND status = 'valid'") (defn- get-database-object [conn id] @@ -213,29 +180,93 @@ (dm/export impl/object?) (defn get-object - [{:keys [::db/connectable] :as storage} id] + [storage id] (assert (valid-storage? storage)) - (get-database-object connectable id)) + (let [ds (db/get-connectable storage)] + (get-database-object ds id))) (defn put-object! "Creates a new object with the provided content." - [{:keys [::backend] :as storage} {:keys [::content] :as params}] + [{:keys [::backend ::db/pool] :as storage} + {:keys [::content ::expired-at ::touched-at ::touch] :as params}] (assert (valid-storage? storage)) (assert (impl/content? content) "expected an instance of content") - (let [object (create-database-object storage params)] - (if (::created? (meta object)) - ;; Store the data finally on the underlying storage subsystem. - (-> (impl/resolve-backend storage backend) - (impl/put-object object content)) - object))) + (let [id (or (::id params) (uuid/random)) + mdata (cond-> (get-metadata params) + (satisfies? impl/IContentHash content) + (assoc :hash (impl/get-hash content))) + + touched-at (if touch + (or touched-at (ct/now)) + touched-at) + + backend' (impl/resolve-backend storage backend)] + + ;; NOTE: for now we don't reuse the deleted objects, but in futute + ;; we can consider reusing deleted objects if we found a duplicated + ;; one and is marked for deletion but still not deleted. + + ;; PHASE 1: deduplication lookup. + (if-some [hit (when (and (::deduplicate? params) + (:hash mdata) + (:bucket mdata) + (not= tempfile-bucket (:bucket mdata))) + (get-database-object-by-hash pool backend + (:bucket mdata) + (:hash mdata)))] + + ;; PHASE 2: an existing reference is found: reuse or repair it. + (if (impl/exists-object? backend' hit) + + ;; PHASE 2a: healthy reference. Optionally refresh touched_at + ;; and reuse the object as it is. + (do + (when touch + (db/update! pool :storage-object + {:touched-at touched-at} + {:id (:id hit)} + {::db/return-keys false})) + (row->storage-object (cond-> hit touch (assoc :touched-at touched-at)))) + + ;; PHASE 2b: the referenced blob is missing (a stale/broken row). + ;; Repair the reference in place: rewrite the incoming content + ;; under the same id, restoring the blob for all existing + ;; references to it. If the write fails, the exception propagates + ;; and the row stays live and valid, so a later matching upload + ;; retries the heal. + (let [object (row->storage-object hit)] + (l/wrn :hint "blob not found on reusing storage object" + :id (:id object) + :backend (name backend)) + (impl/put-object backend' object content) + (promote-object! storage object) + object)) + + ;; PHASE 3: no dedup hit: create a fresh object. The row is + ;; inserted in 'pending' state so it is not visible to the normal + ;; lifecycle (dedup, gc, reads) until the blob has been written + ;; and the object promoted to 'valid'. + (let [row (db/insert! pool :storage-object + {:id id + :size (impl/get-size content) + :backend (name backend) + :metadata (db/tjson mdata) + :deleted-at expired-at + :touched-at touched-at + :status "pending"}) + object (row->storage-object row)] + (impl/put-object backend' object content) + (promote-object! storage object) + object)))) (defn touch-object! "Mark object as touched." - [{:keys [::db/connectable] :as storage} object-or-id] + [storage object-or-id] (assert (valid-storage? storage)) - (let [id (if (impl/object? object-or-id) (:id object-or-id) object-or-id)] - (-> (db/update! connectable :storage-object + (let [id (if (impl/object? object-or-id) (:id object-or-id) object-or-id) + ds (db/get-connectable storage)] + (-> (db/update! ds :storage-object {:touched-at (ct/now)} {:id id}) (db/get-update-count) @@ -282,10 +313,11 @@ (-> (impl/get-object-url backend object nil) file-url->path)))) (defn del-object! - [{:keys [::db/connectable] :as storage} object-or-id] + [storage object-or-id] (assert (valid-storage? storage)) (let [id (if (impl/object? object-or-id) (:id object-or-id) object-or-id) - res (db/update! connectable :storage-object + ds (db/get-connectable storage) + res (db/update! ds :storage-object {:deleted-at (ct/now)} {:id id})] (pos? (db/get-update-count res)))) @@ -295,9 +327,10 @@ (dm/export impl/get-size) (defn configure - [storage connectable] + [storage connection] + (assert (db/connection? connection)) (assert (valid-storage? storage)) - (assoc storage ::db/connectable connectable)) + (assoc storage ::db/conn connection)) (defn resolve "Resolves the storage instance with preconfigured backend. You can @@ -306,5 +339,5 @@ [cfg & {:as opts}] (let [storage (::storage cfg)] (if (::db/reuse-conn opts false) - (configure storage (db/get-connectable cfg)) + (configure storage (db/get-connection cfg)) storage))) diff --git a/backend/src/app/storage/fs.clj b/backend/src/app/storage/fs.clj index a90fecff88..cf07128579 100644 --- a/backend/src/app/storage/fs.clj +++ b/backend/src/app/storage/fs.clj @@ -11,6 +11,7 @@ [app.common.uri :as u] [app.storage :as-alias sto] [app.storage.impl :as impl] + [app.storage.tmp :as tmp] [cuerdas.core :as str] [datoteka.fs :as fs] [datoteka.io :as io] @@ -18,8 +19,11 @@ (:import java.io.InputStream java.io.OutputStream + java.nio.file.AtomicMoveNotSupportedException + java.nio.file.CopyOption java.nio.file.Files - java.nio.file.Path)) + java.nio.file.Path + java.nio.file.StandardCopyOption)) (set! *warn-on-reflection* true) @@ -59,17 +63,49 @@ (assert (valid-backend? backend) "expected a valid backend instance") (let [base (fs/path (::directory backend)) path (fs/path (impl/id->path id)) - full (fs/normalize (fs/join base path))] + full (fs/normalize (fs/join base path)) + parent-dir (fs/parent full)] - (when-not (fs/exists? (fs/parent full)) - (fs/create-dir (fs/parent full))) + (when-not (fs/exists? parent-dir) + (fs/create-dir parent-dir)) - (with-open [^InputStream src (io/input-stream content)] - (with-open [^OutputStream dst (io/output-stream full)] - (io/copy src dst))) + ;; Create temp file in the same directory (same filesystem → atomic + ;; move preserved) and register with cleanup queue (crashed-JVM files + ;; are swept ~60min later). + (let [tmp (tmp/tempfile :dir (str parent-dir) + :prefix (str (fs/name full) ".") + :suffix ".tmp" + :min-age "1h")] + + ;; Write to a temporary file in the same directory and atomically move + ;; it into place, so a failed write never leaves a partial blob at the + ;; final path. + (try + (with-open [^InputStream src (io/input-stream content)] + (with-open [^OutputStream dst (io/output-stream tmp)] + (io/copy src dst))) + ;; ATOMIC_MOVE is POSIX-only; on non-POSIX filesystems (e.g. Windows) + ;; this may throw FileAlreadyExistsException if the target exists. + (try + (Files/move ^Path tmp ^Path full + (into-array CopyOption [StandardCopyOption/ATOMIC_MOVE])) + (catch AtomicMoveNotSupportedException _ + (Files/move ^Path tmp ^Path full + (into-array CopyOption [StandardCopyOption/REPLACE_EXISTING])))) + (catch Throwable cause + ;; Temp file cleanup is handled by the cleanup queue; just rethrow. + (throw cause)))) object)) +(defmethod impl/exists-object? :fs + [backend {:keys [id]}] + (assert (valid-backend? backend) "expected a valid backend instance") + (let [^Path base (fs/path (::directory backend)) + ^Path path (fs/path (impl/id->path id)) + ^Path full (fs/normalize (fs/join base path))] + (fs/exists? full))) + (defmethod impl/get-object-data :fs [backend {:keys [id] :as object}] (assert (valid-backend? backend) "expected a valid backend instance") @@ -108,8 +144,12 @@ [backend ids] (assert (valid-backend? backend) "expected a valid backend instance") (let [base (fs/path (::directory backend))] - (doseq [id ids] - (let [path (fs/path (impl/id->path id)) - path (fs/join base path)] - (Files/deleteIfExists ^Path path))))) + (reduce (fn [fail-ids id] + (let [path (fs/normalize (fs/join base (fs/path (impl/id->path id))))] + (try + (Files/deleteIfExists ^Path path) + fail-ids + (catch Throwable _ + (conj fail-ids id))))) + #{} ids))) diff --git a/backend/src/app/storage/gc_deleted.clj b/backend/src/app/storage/gc_deleted.clj index 2aadca4b55..ab49030041 100644 --- a/backend/src/app/storage/gc_deleted.clj +++ b/backend/src/app/storage/gc_deleted.clj @@ -19,8 +19,18 @@ [app.db :as db] [app.storage :as sto] [app.storage.impl :as impl] + [clojure.set :as set] [integrant.core :as ig])) +(def ^:private max-attempts + "Maximum number of deletion attempts before giving up and accepting + the orphan blob." + 7) + +(def ^:private chunk-size + "Number of rows to process per transaction." + 25) + (def ^:private sql:lock-sobjects "SELECT id FROM storage_object WHERE id = ANY(?::uuid[]) @@ -47,66 +57,110 @@ (-> (db/exec-one! conn [sql:delete-sobjects ids]) (db/get-update-count)))) -(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. - (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)) +(def ^:private sql:increment-attempts-and-defer + "UPDATE storage_object + SET deletion_attempts = deletion_attempts + 1, + deleted_at = NOW() + INTERVAL '1 day' + WHERE id = ANY(?::uuid[])") - (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- increment-attempts-and-defer! + [conn ids] + (let [ids (db/create-array conn "uuid" ids)] + (db/exec-one! conn [sql:increment-attempts-and-defer ids]))) +(def ^:private sql:delete-give-up + "DELETE FROM storage_object + WHERE id = ANY(?::uuid[]) + AND deletion_attempts >= ?") + +(defn- delete-give-up! + [conn ids] + (let [ids (db/create-array conn "uuid" ids)] + (db/exec-one! conn [sql:delete-give-up ids max-attempts]))) + +(defn- process-chunk + "Attempt to delete a chunk of storage objects from a specific backend. + + This function runs inside the caller's transaction (clean-deleted!) — + it does NOT open its own transaction. The caller is responsible for + ensuring the rows are locked via FOR UPDATE SKIP LOCKED before calling. + + Returns the number of successfully deleted objects, or 0 if no rows + could be locked." + [conn storage backend-id ids] + (if-let [locked-ids (lock-ids conn ids)] + (let [fail-ids (try + (-> (impl/resolve-backend storage backend-id) + (impl/del-objects-in-bulk locked-ids)) + (catch Throwable cause + (l/err :hint "error on physical deletion, will retry" + :ids locked-ids + :cause cause) + locked-ids)) + ok-ids (set/difference locked-ids fail-ids)] + + (doseq [id ok-ids] + (l/dbg :hint "permanently delete storage object" + :id (str id) + :backend (name backend-id))) + + (when (seq ok-ids) + (delete-sobjects! conn ok-ids)) + + (when (seq fail-ids) + (increment-attempts-and-defer! 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" + :ids fail-ids + :max-attempts max-attempts)))) + + (count ok-ids)) + 0)) (defn- group-by-backend [items] (d/group-by (comp keyword :backend) :id #{} items)) -(def ^:private sql:get-deleted-sobjects - "SELECT s.* - FROM storage_object AS s - WHERE s.deleted_at IS NOT NULL - AND s.deleted_at <= ? - ORDER BY s.deleted_at ASC") +(def ^:private sql:get-deleted-chunk + "SELECT id, backend + FROM storage_object + WHERE deleted_at IS NOT NULL + AND deleted_at <= ? + AND status = 'valid' + ORDER BY deleted_at ASC + LIMIT ? + FOR UPDATE + SKIP LOCKED") -(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])))) +(defn- get-deleted-chunk + [conn size] + (db/exec! conn [sql:get-deleted-chunk (ct/now) size])) (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))) + [cfg] + (loop [total 0] + (let [deleted (db/tx-run! cfg + (fn [{:keys [::db/conn ::sto/storage]}] + (let [chunk (get-deleted-chunk conn chunk-size)] + (when (seq chunk) + (let [by-backend (group-by-backend chunk)] + (reduce-kv (fn [acc backend-id ids] + (+ acc (process-chunk conn storage backend-id ids))) + 0 + by-backend))))))] + (if deleted + (recur (+ total deleted)) + 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 db pool")) (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/src/app/storage/gc_touched.clj b/backend/src/app/storage/gc_touched.clj index cfdf6788f7..af16af5f82 100644 --- a/backend/src/app/storage/gc_touched.clj +++ b/backend/src/app/storage/gc_touched.clj @@ -23,7 +23,6 @@ [app.common.exceptions :as ex] [app.common.logging :as l] [app.common.time :as ct] - [app.config :as cf] [app.db :as db] [app.storage :as sto] [app.storage.impl :as impl] @@ -108,10 +107,9 @@ WHERE id = ANY(?::uuid[])") (defn- mark-delete-in-bulk! - [conn deletion-delay ids] - (let [ids (db/create-array conn "uuid" ids) - now (ct/plus (ct/now) deletion-delay)] - (db/exec-one! conn [sql:mark-delete-in-bulk now ids]))) + [conn ids] + (let [ids (db/create-array conn "uuid" ids)] + (db/exec-one! conn [sql:mark-delete-in-bulk (ct/now) ids]))) ;; NOTE: A getter that retrieves the key which will be used for group ;; ids; previously we have no value, then we introduced the @@ -149,11 +147,9 @@ :status "delete" :bucket bucket) (recur to-freeze (conj to-delete id) (rest objects)))) - (let [deletion-delay (if (= sto/tempfile-bucket bucket) - (ct/duration {:hours 2}) - (cf/get-deletion-delay))] + (do (some->> (seq to-freeze) (mark-freeze-in-bulk! conn)) - (some->> (seq to-delete) (mark-delete-in-bulk! conn deletion-delay)) + (some->> (seq to-delete) (mark-delete-in-bulk! conn)) [(count to-freeze) (count to-delete)])))) (defn- process-bucket! @@ -186,6 +182,7 @@ FROM storage_object AS so WHERE so.touched_at IS NOT NULL AND so.touched_at <= ? + AND so.status = 'valid' ORDER BY touched_at ASC FOR UPDATE SKIP LOCKED @@ -221,7 +218,9 @@ (defmethod ig/init-key ::handler [_ {:keys [::min-age] :as cfg}] - (fn [_] - (let [threshold (ct/minus (ct/now) min-age)] + (fn [{:keys [props]}] + (let [threshold (if (:skip-delay props) + (ct/now) + (ct/minus (ct/now) min-age))] (process-touched! (assoc cfg ::timestamp threshold))))) diff --git a/backend/src/app/storage/impl.clj b/backend/src/app/storage/impl.clj index 9b5a33b7dd..76fafe2cac 100644 --- a/backend/src/app/storage/impl.clj +++ b/backend/src/app/storage/impl.clj @@ -71,7 +71,10 @@ :code :invalid-storage-backend :context cfg)) -(defmulti del-objects-in-bulk (fn [cfg _] (::sto/type cfg))) +(defmulti del-objects-in-bulk + "Delete multiple objects in bulk. Returns #{fail-ids} — the set of ids + whose blob deletion failed. Empty set = all succeeded." + (fn [cfg _] (::sto/type cfg))) (defmethod del-objects-in-bulk :default [cfg _] @@ -79,6 +82,14 @@ :code :invalid-storage-backend :context cfg)) +(defmulti exists-object? (fn [cfg _] (::sto/type cfg))) + +(defmethod exists-object? :default + [cfg _] + (ex/raise :type :internal + :code :invalid-storage-backend + :context cfg)) + ;; --- HELPERS (defn uuid->hex diff --git a/backend/src/app/storage/pending_gc.clj b/backend/src/app/storage/pending_gc.clj new file mode 100644 index 0000000000..7b8f04ff13 --- /dev/null +++ b/backend/src/app/storage/pending_gc.clj @@ -0,0 +1,88 @@ +;; 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 app.storage.pending-gc + "A maintenance task that reclaims storage objects created in 'pending' + state that were never promoted to 'valid' (e.g. after a crash between + writing the blob and promoting the row). + + Pending rows are invisible to the normal lifecycle (dedup, gc, reads). This + task removes the orphaned blob (if any) and the pending row itself, without + ever iterating the whole physical store." + (:require + [app.common.logging :as l] + [app.db :as db] + [app.storage :as sto] + [app.storage.impl :as impl] + [integrant.core :as ig])) + +(def ^:private sql:get-pending-sobjects + "SELECT id, backend + FROM storage_object + WHERE status = 'pending' + AND created_at <= now() - interval '24 hours' + ORDER BY created_at ASC + LIMIT ? + FOR UPDATE + SKIP LOCKED") + +(defn- get-pending-chunk + [conn chunk-size] + (db/exec! conn [sql:get-pending-sobjects chunk-size])) + +(def ^:private sql:delete-pending-sobject + "DELETE FROM storage_object WHERE id = ? AND status = 'pending'") + +(def ^:private chunk-size + 100) + +(defn- delete-pending-rows! + "Select, lock and delete a chunk of pending rows in a single transaction. + Returns the deleted rows or nil when there is nothing left to reclaim." + [cfg] + (db/tx-run! cfg + (fn [{:keys [::db/conn]}] + ;; NOTE: db/exec! returns an empty vector when there are no + ;; rows left; use not-empty to detect it. + (when-let [chunk (not-empty (get-pending-chunk conn chunk-size))] + (doseq [{:keys [id]} chunk] + (db/exec-one! conn [sql:delete-pending-sobject id])) + chunk)))) + +(defn- delete-blobs! + "Best-effort removal of the orphaned blobs. Runs after the pending rows + have been committed so a failure here never blocks their reclamation." + [storage rows] + (doseq [{:keys [id backend]} rows] + (try + (-> (impl/resolve-backend storage (keyword backend)) + (impl/del-object {:id id})) + (catch Throwable cause + (l/err :hint "error deleting orphaned pending blob" + :id (str id) + :backend backend + :cause cause))))) + +(defn- process! + [{::sto/keys [storage] :as cfg}] + (loop [total 0] + (if-let [rows (delete-pending-rows! cfg)] + (do + (delete-blobs! storage rows) + (recur (long (+ total (count rows))))) + total))) + +(defmethod ig/assert-key ::handler + [_ params] + (assert (db/pool? (::db/pool params)) "expected valid db pool") + (assert (sto/valid-storage? (::sto/storage params)) "expect valid storage")) + +(defmethod ig/init-key ::handler + [_ cfg] + (fn [_] + (let [total (process! cfg)] + (l/inf :hint "task finished" :total total) + {:processed total}))) diff --git a/backend/src/app/storage/s3.clj b/backend/src/app/storage/s3.clj index c55158a7ba..6cf97321f1 100644 --- a/backend/src/app/storage/s3.clj +++ b/backend/src/app/storage/s3.clj @@ -47,6 +47,7 @@ software.amazon.awssdk.services.s3.model.DeleteObjectsRequest software.amazon.awssdk.services.s3.model.DeleteObjectsResponse software.amazon.awssdk.services.s3.model.GetObjectRequest + software.amazon.awssdk.services.s3.model.HeadObjectRequest software.amazon.awssdk.services.s3.model.NoSuchKeyException software.amazon.awssdk.services.s3.model.ObjectIdentifier software.amazon.awssdk.services.s3.model.PutObjectRequest @@ -78,6 +79,7 @@ (declare get-object-url) (declare del-object) (declare del-object-in-bulk) +(declare head-object) (declare build-s3-client) (declare build-s3-presigner) @@ -186,10 +188,46 @@ [backend object] (p/await! (del-object backend object))) +(defmethod impl/exists-object? :s3 + [backend object] + (assert (valid-backend? backend) "expected a valid backend instance") + (loop [result (p/await (head-object backend object)) + retryn 0] + (if (ex/exception? result) + (cond + ;; A missing key is a definitive answer, no need to retry. + (ex/instance? NoSuchKeyException result) + false + + ;; Any other error is considered transient and retried. + (< retryn max-retries) + (do + (Thread/sleep (* 100 (inc retryn))) + (recur (p/await (head-object backend object)) (inc retryn))) + + :else + (throw result)) + true))) + (defmethod impl/del-objects-in-bulk :s3 [backend ids] (assert (valid-backend? backend) "expected a valid backend instance") - (p/await! (del-object-in-bulk backend ids))) + (let [key->id (into {} (map (fn [id] + [(str (::prefix backend) (impl/id->path id)) id])) + ids) + result (try + (p/await! (del-object-in-bulk backend ids)) + (catch Throwable cause + (l/err :hint "error on s3 bulk deletion" + :ids ids + :cause cause) + ::network-error))] + (cond + (= ::network-error result) (set ids) + (map? result) (into #{} (map (fn [{:keys [key]}] + (get key->id key))) + (:errors result)) + :else #{}))) ;; --- HELPERS @@ -330,6 +368,14 @@ ^AsyncResponseTransformer rxf) (p/fmap #(.asInputStream ^ResponseBytes %))))))) +(defn- head-object + [{:keys [::client ::bucket ::prefix]} {:keys [id]}] + (let [hor (.. (HeadObjectRequest/builder) + (bucket bucket) + (key (str prefix (impl/id->path id))) + (build))] + (.headObject ^S3AsyncClient client ^HeadObjectRequest hor))) + (defn- get-object-bytes [{:keys [::client ::bucket ::prefix]} {:keys [id]}] (let [gor (.. (GetObjectRequest/builder) @@ -379,12 +425,11 @@ (defn- del-object-in-bulk [{:keys [::bucket ::client ::prefix]} ids] - - (let [oids (map (fn [id] - (.. (ObjectIdentifier/builder) - (key (str prefix (impl/id->path id))) - (build))) - ids) + (let [oids (mapv (fn [id] + (.. (ObjectIdentifier/builder) + (key (str prefix (impl/id->path id))) + (build))) + ids) delc (.. (Delete/builder) (objects ^Collection oids) (build)) @@ -392,14 +437,9 @@ (bucket bucket) (delete ^Delete delc) (build))] - (->> (.deleteObjects ^S3AsyncClient client ^DeleteObjectsRequest dor) - (p/fmap (fn [dres] - (when (.hasErrors ^DeleteObjectsResponse dres) - (let [errors (seq (.errors ^DeleteObjectsResponse dres))] - (ex/raise :type :internal - :code :error-on-s3-bulk-delete - :s3-errors (mapv (fn [^S3Error error] - {:key (.key error) - :msg (.message error)}) - errors))))))))) + (p/fmap (fn [^DeleteObjectsResponse dres] + (when (.hasErrors dres) + {:errors (mapv (fn [^S3Error e] + {:key (.key e) :msg (.message e)}) + (.errors dres))})))))) diff --git a/backend/src/app/storage/tmp.clj b/backend/src/app/storage/tmp.clj index 4448920d99..7a4ae3b901 100644 --- a/backend/src/app/storage/tmp.clj +++ b/backend/src/app/storage/tmp.clj @@ -80,11 +80,12 @@ ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; (defn tempfile* - [& {:keys [suffix prefix] + [& {:keys [suffix prefix dir] :or {prefix "penpot." - suffix ".tmp"}}] + suffix ".tmp" + dir default-tmp-dir}}] (let [attrs (fs/make-permissions "rw-r--r--") - path (fs/join default-tmp-dir (str prefix (uuid/next) suffix))] + path (fs/join dir (str prefix (uuid/next) suffix))] (Files/createFile path attrs))) (defn tempfile diff --git a/backend/src/app/tasks/file_gc.clj b/backend/src/app/tasks/file_gc.clj index f2e6ee16b5..78461e9b99 100644 --- a/backend/src/app/tasks/file_gc.clj +++ b/backend/src/app/tasks/file_gc.clj @@ -251,12 +251,9 @@ (try (-> cfg (assoc ::db/rollback (:rollback? props)) - (db/tx-run! (fn [{:keys [::db/conn] :as cfg}] - (let [cfg (-> cfg - (update ::sto/storage sto/configure conn) - (assoc ::timestamp (ct/now))) - processed? (process-file! cfg props)] - + (assoc ::timestamp (ct/now)) + (db/tx-run! (fn [cfg] + (let [processed? (process-file! cfg props)] (when (and processed? (contains? cf/flags :tiered-file-data-storage)) (wrk/submit! (-> cfg (assoc ::wrk/task :offload-file-data) diff --git a/backend/src/app/tasks/objects_gc.clj b/backend/src/app/tasks/objects_gc.clj index 8b6c90e82d..a72459e432 100644 --- a/backend/src/app/tasks/objects_gc.clj +++ b/backend/src/app/tasks/objects_gc.clj @@ -321,8 +321,14 @@ (defmethod ig/init-key ::handler [_ cfg] - (fn [_] - (let [cfg (assoc cfg ::timestamp (ct/now))] + (fn [{:keys [props]}] + (let [skip-delay (:skip-delay props) + chunk-size (or (:chunk-size props) (::chunk-size cfg)) + cfg (-> cfg + (assoc ::chunk-size chunk-size) + (assoc ::timestamp (if skip-delay + (ct/in-future {:days 3650}) + (ct/now))))] (loop [procs (map deref deletion-proc-vars) total 0] (if-let [proc-fn (first procs)] diff --git a/backend/test/backend_tests/storage_test.clj b/backend/test/backend_tests/storage_test.clj index adc84e64ac..cc34773e67 100644 --- a/backend/test/backend_tests/storage_test.clj +++ b/backend/test/backend_tests/storage_test.clj @@ -12,12 +12,23 @@ [app.db :as db] [app.rpc :as-alias rpc] [app.storage :as sto] + [app.storage.fs :as-alias sto.fs] + [app.storage.impl :as impl] + [app.storage.s3 :as-alias sto.s3] [backend-tests.helpers :as th] [clojure.test :as t] [cuerdas.core :as str] [datoteka.fs :as fs] [datoteka.io :as io] - [mockery.core :refer [with-mocks]])) + [mockery.core :refer [with-mocks]] + [promesa.core :as p]) + (:import + (software.amazon.awssdk.services.s3 + S3AsyncClient) + (software.amazon.awssdk.services.s3.model + NoSuchKeyException) + (software.amazon.awssdk.services.s3.presigner + S3Presigner))) (t/use-fixtures :once th/state-init) (t/use-fixtures :each (th/serial @@ -368,27 +379,498 @@ now (ct/now) object1 (sto/put-object! storage {::sto/content content1 - ::sto/touched-at (ct/plus now {:minutes 1}) + ::sto/touched-at (ct/plus now {:hours 1}) :bucket "tempfile" :content-type "text/plain"})] - + ;; not eligible while the touched-at is in the future (binding [ct/*clock* (ct/fixed-clock now)] (let [res (th/run-task! :storage-gc-touched {})] (t/is (= 0 (:freeze res))) (t/is (= 0 (:delete res))))) - - (binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 3}))] + ;; still not eligible: touched-at (now+1h) is beyond the threshold + (binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 2}))] (let [res (th/run-task! :storage-gc-touched {})] + (t/is (= 0 (:freeze res))) + (t/is (= 0 (:delete res))))) + + ;; eligible: marked for deletion immediately, without any extra delay + (let [clock (ct/plus now {:hours 3})] + (binding [ct/*clock* (ct/fixed-clock clock)] + (let [res (th/run-task! :storage-gc-touched {})] + (t/is (= 0 (:freeze res))) + (t/is (= 1 (:delete res))))) + + (let [row (th/db-exec-one! ["select deleted_at from storage_object where id = ?" (:id object1)])] + (t/is (ct/is-before-or-equal? (:deleted-at row) (ct/plus clock {:seconds 1}))))) + + ;; removed on the next deleted gc run + (binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 4}))] + (let [res (th/run-task! :storage-gc-deleted {})] + (t/is (= 1 (:deleted res))))))) + +(t/deftest touched-gc-task-skip-delay + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content1") + now (ct/now) + + object1 (sto/put-object! storage {::sto/content content + ::sto/touched-at now + :bucket "tempfile" + :content-type "text/plain"})] + + ;; too recent: not processed without skip-delay + (binding [ct/*clock* (ct/fixed-clock now)] + (let [res (th/run-task! :storage-gc-touched {})] + (t/is (= 0 (:freeze res))) + (t/is (= 0 (:delete res))))) + + ;; processed immediately with skip-delay + (binding [ct/*clock* (ct/fixed-clock now)] + (let [res (th/run-task! :storage-gc-touched {:skip-delay true})] (t/is (= 0 (:freeze res))) (t/is (= 1 (:delete res))))) + ;; and marked for deletion without any additional delay + (let [row (th/db-exec-one! ["select deleted_at from storage_object where id = ?" (:id object1)])] + (t/is (ct/is-before-or-equal? (:deleted-at row) (ct/plus now {:seconds 1})))))) - (binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 1}))] +(t/deftest storage-gc-deleted-immediate + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content1") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"})] + + ;; mark as deleted right now + (th/db-exec! ["update storage_object set deleted_at = ?" (ct/now)]) + + ;; the deleted gc removes it on the next run + (let [res (th/run-task! :storage-gc-deleted {})] + (t/is (= 1 (:deleted res)))))) + +(t/deftest objects-gc-task-skip-delay + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + prof (th/create-profile* 1) + proj (th/create-project* 1 {:profile-id (:id prof) + :team-id (:default-team-id prof)}) + file (th/create-file* 1 {:profile-id (:id prof) + :project-id (:default-project-id prof) + :is-shared false}) + mfile {:filename "sample.jpg" + :path (th/tempfile "backend_tests/test_files/sample.jpg") + :mtype "image/jpeg" + :size 312043} + params {::th/type :upload-file-media-object + ::rpc/profile-id (:id prof) + :file-id (:id file) + :is-local true + :name "testfile" + :content mfile} + out1 (th/command! params) + out2 (th/command! params)] + + (t/is (nil? (:error out1))) + (t/is (nil? (:error out2))) + + (let [result-1 (:result out1) + result-2 (:result out2)] + + ;; mark as deleted but in the future (not yet eligible) + (th/db-update! :file-media-object + {:deleted-at (ct/in-future {:days 1})} + {:id (:id result-1)}) + + ;; without skip-delay the future deleted row is not processed + (let [res (th/run-task! :objects-gc {})] + (t/is (= 0 (:processed res)))) + + ;; with skip-delay it is processed immediately + (let [res (th/run-task! :objects-gc {:skip-delay true})] + (t/is (= 1 (:processed res))))))) + +(t/deftest put-object-write-failure-leaves-pending-row + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + ;; Point the fs backend at a path that is actually a file so the + ;; blob write fails. + blocked (fs/path "/tmp/penpot" (str "blocked-" (uuid/next))) + _ (spit (str blocked) "x")] + (try + (let [broken (assoc-in storage [::sto/backends :fs ::sto.fs/directory] (str blocked)) + content (sto/content "content") + ex (try + (sto/put-object! broken {::sto/content content + :content-type "text/plain"}) + nil + (catch Throwable cause cause))] + (t/is (some? ex)) + + ;; the pending row stays behind and is reclaimed asynchronously + ;; by the :storage-pending-gc task + (let [rows (th/db-query :storage-object {:status "pending"})] + (t/is (= 1 (count rows))) + + (th/db-update! :storage-object + {:created-at (ct/in-past {:days 2})} + {:id (:id (first rows))}) + + (let [res (th/run-task! :storage-pending-gc {})] + (t/is (= 1 (:processed res)))) + + (let [row (th/db-exec-one! ["select count(*) from storage_object"])] + (t/is (= 0 (:count row)))))) + (finally + (fs/delete blocked))))) + +(t/deftest pending-gc-reclaims-unpromoted-object + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"}) + path (sto/get-object-path storage object)] + + ;; valid objects are never reclaimed + (let [res (th/run-task! :storage-pending-gc {})] + (t/is (= 0 (:processed res)))) + + ;; simulate a crash: the object was created but never promoted + (th/db-update! :storage-object {:status "pending" + :created-at (ct/in-past {:days 2})} + {:id (:id object)}) + + (t/is (fs/exists? path)) + + (let [res (th/run-task! :storage-pending-gc {})] + (t/is (= 1 (:processed res)))) + + ;; both the row and the orphaned blob are removed + (let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])] + (t/is (= 0 (:count row)))) + (t/is (not (fs/exists? path))))) + +(t/deftest pending-objects-excluded-from-gc-touched + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + ::sto/touched-at (ct/now) + :content-type "text/plain"})] + + ;; mark it pending and touched in the past + (th/db-update! :storage-object {:status "pending" + :touched-at (ct/in-past {:days 1})} + {:id (:id object)}) + + (binding [ct/*clock* (ct/fixed-clock (ct/now))] + (let [res (th/run-task! :storage-gc-touched {})] + (t/is (= 0 (:freeze res))) + (t/is (= 0 (:delete res))))) + + ;; still present and not marked as deleted + (let [row (th/db-exec-one! ["select * from storage_object where id = ?" (:id object)])] + (t/is (some? row)) + (t/is (nil? (:deleted-at row)))))) + +(t/deftest pending-objects-excluded-from-get + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"})] + + (t/is (some? (sto/get-object storage (:id object)))) + + (th/db-update! :storage-object {:status "pending"} {:id (:id object)}) + + (t/is (nil? (sto/get-object storage (:id object)))))) + +(t/deftest pending-objects-excluded-from-dedup + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (-> (sto/content "content") + (sto/wrap-with-hash "same-hash")) + object1 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"})] + + ;; mark the only matching row as pending + (th/db-update! :storage-object {:status "pending"} {:id (:id object1)}) + + (let [object2 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"})] + (t/is (not= (:id object1) (:id object2)))))) + +(t/deftest dedup-reuses-existing-blob + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (-> (sto/content "content") + (sto/wrap-with-hash "same-hash")) + object1 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"}) + object2 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"})] + (t/is (= (:id object1) (:id object2))) + (let [row (th/db-exec-one! ["select count(*) from storage_object"])] + (t/is (= 1 (:count row)))))) + +(t/deftest dedup-repairs-stale-object + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (-> (sto/content "content") + (sto/wrap-with-hash "same-hash")) + object1 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"})] + + ;; remove the physical blob to simulate a stale/broken object + (let [path (sto/get-object-path storage object1)] + (fs/delete path)) + + ;; re-uploading identical content repairs the same reference in place + (let [object2 (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"})] + (t/is (= (:id object1) (:id object2))) + + ;; the row stays live: no tombstone and no extra row + (let [row (th/db-exec-one! ["select status, deleted_at from storage_object where id = ?" (:id object1)])] + (t/is (= "valid" (:status row))) + (t/is (nil? (:deleted-at row)))) + + (let [row (th/db-exec-one! ["select count(*) from storage_object"])] + (t/is (= 1 (:count row)))) + + ;; the repaired blob is readable again under the original id + (t/is (= "content" (slurp (sto/get-object-data storage object2))))))) + +(t/deftest gc-deleted-removes-broken-object + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"})] + + ;; mark as deleted and remove the physical blob + (th/db-update! :storage-object {:deleted-at (ct/in-past {:minutes 1})} + {:id (:id object)}) + (let [path (sto/get-object-path storage object)] + (fs/delete path)) + + ;; the deleted gc removes the row without error even though the blob is + ;; missing (the physical deletion is best-effort) + (let [res (th/run-task! :storage-gc-deleted {})] + (t/is (= 1 (:deleted res)))) + + (let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])] + (t/is (= 0 (:count row)))))) + +(t/deftest pending-objects-excluded-from-gc-deleted + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"})] + ;; mark as pending + deleted in the past + (th/db-update! :storage-object {:status "pending" + :deleted-at (ct/in-past {:minutes 1})} + {:id (:id object)}) + ;; gc-deleted skips it because status != 'valid' + (let [res (th/run-task! :storage-gc-deleted {})] + (t/is (= 0 (:deleted res)))) + ;; row still exists (with deleted_at set — we set it above) + (let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" + (:id object)])] + (t/is (= 1 (:count row)))))) + +(t/deftest gc-deleted-gives-up-after-max-attempts + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (sto/content "content") + object (sto/put-object! storage {::sto/content content + :content-type "text/plain"})] + + (th/db-update! :storage-object {:deleted-at (ct/in-past {:minutes 1}) + :deletion_attempts 6} + {:id (:id object)}) + + (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))))) - (binding [ct/*clock* (ct/fixed-clock (ct/plus now {:hours 2}))] - (let [res (th/run-task! :storage-gc-deleted {})] - (t/is (= 0 (:deleted res))))))) + (let [row (th/db-exec-one! ["select count(*) from storage_object where id = ?" (:id object)])] + (t/is (= 0 (:count row)))))) + +(t/deftest dedup-reuses-existing-blob-with-touch + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (-> (sto/content "content") + (sto/wrap-with-hash "same-hash")) + t0 (ct/now) + params {::sto/deduplicate? true + ::sto/touch true + :bucket "file-media-object" + :content-type "text/plain"} + object1 (binding [ct/*clock* (ct/fixed-clock t0)] + (sto/put-object! storage (assoc params ::sto/content content)))] + + ;; a touched hit reuses the object and updates its touched_at + (let [object2 (binding [ct/*clock* (ct/fixed-clock (ct/plus t0 {:hours 1}))] + (sto/put-object! storage (assoc params ::sto/content content)))] + (t/is (= (:id object1) (:id object2))) + + (let [row (th/db-exec-one! ["select touched_at from storage_object where id = ?" (:id object1)])] + (t/is (ct/is-after? (:touched-at row) t0)))) + + ;; with the blob removed, the touched hit repairs the stale row in + ;; place: the same id is kept, the row is not deleted and touched_at + ;; is left untouched (the touch flag only applies to healthy hits) + (let [path (sto/get-object-path storage object1)] + (fs/delete path)) + + (let [object3 (binding [ct/*clock* (ct/fixed-clock (ct/plus t0 {:hours 2}))] + (sto/put-object! storage (assoc params ::sto/content content)))] + (t/is (= (:id object1) (:id object3))) + + (let [row (th/db-exec-one! ["select deleted_at, touched_at from storage_object where id = ?" (:id object1)])] + (t/is (nil? (:deleted-at row))) + ;; the touch flag does not apply to repairs: touched_at was last + ;; set by the healthy hit and is not bumped by the repair + (t/is (ct/is-before? (:touched-at row) (ct/plus t0 {:hours 2}))))))) + +(t/deftest put-object-repair-failure-leaves-row-intact + (let [storage (-> (:app.storage/storage th/*system*) + (configure-storage-backend)) + content (-> (sto/content "content") + (sto/wrap-with-hash "same-hash")) + object (sto/put-object! storage {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"}) + path (sto/get-object-path storage object) + + ;; Point the fs backend at a path that is actually a file so the + ;; blob write fails. + blocked (fs/path "/tmp/penpot" (str "blocked-" (uuid/next))) + _ (spit (str blocked) "x")] + (try + ;; remove the physical blob to simulate a stale/broken object + (fs/delete path) + + (let [broken (assoc-in storage [::sto/backends :fs ::sto.fs/directory] (str blocked)) + ex (try + (sto/put-object! broken {::sto/content content + ::sto/deduplicate? true + :bucket "file-media-object" + :content-type "text/plain"}) + nil + (catch Throwable cause cause))] + (t/is (some? ex)) + + ;; the failed repair leaves the original row exactly as it was: + ;; live and valid, so a later upload can retry the healing + (let [row (th/db-exec-one! ["select status, deleted_at from storage_object where id = ?" (:id object)])] + (t/is (= "valid" (:status row))) + (t/is (nil? (:deleted-at row)))) + + (let [row (th/db-exec-one! ["select count(*) from storage_object"])] + (t/is (= 1 (:count row))))) + (finally + (fs/delete blocked))))) + +(t/deftest upload-chunks-exclude-pending + (let [prof (th/create-profile* 1) + _ (th/create-project* 1 {:profile-id (:id prof) + :team-id (:default-team-id prof)}) + file (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))) + + ;; mark all the chunks of this session as pending (simulates rows that + ;; were never promoted) + (th/db-exec! ["update storage_object set status = 'pending' where (metadata->>'~:upload-id') = ?" + (str session-id)]) + + ;; assembling fails because no chunk is visible anymore + (let [assemble-out (th/command! {::th/type :assemble-file-media-object + ::rpc/profile-id (:id prof) + :session-id session-id + :file-id (:id file) + :is-local true + :name "assembled-image" + :mtype "image/jpeg"})] + (t/is (some? (:error assemble-out)))))) + +(defn- fake-s3-backend + [] + {::sto/type :s3 + ::sto.s3/client (reify S3AsyncClient) + ::sto.s3/presigner (reify S3Presigner)}) + +(t/deftest s3-exists-object-returns-true-on-found + (with-mocks [mock {:target 'app.storage.s3/head-object + :return (p/resolved {})}] + (t/is (true? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)}))) + (t/is (= 1 (:call-count @mock))))) + +(t/deftest s3-exists-object-returns-false-on-missing-key + (with-mocks [mock {:target 'app.storage.s3/head-object + :return (p/rejected (-> (NoSuchKeyException/builder) + (.message "no key") + (.build)))}] + (t/is (false? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)}))) + ;; a missing key is a definitive answer: no retries + (t/is (= 1 (:call-count @mock))))) + +(t/deftest s3-exists-object-retries-transient-errors + (let [calls (atom 0)] + (with-mocks [_mock {:target 'app.storage.s3/head-object + :return (fn [& _] + (swap! calls inc) + (if (< @calls 3) + (p/rejected (RuntimeException. "boom")) + (p/resolved {})))}] + (t/is (true? (impl/exists-object? (fake-s3-backend) {:id (uuid/next)}))) + (t/is (= 3 @calls))))) + +(t/deftest s3-exists-object-throws-after-retries-exhausted + (with-mocks [mock {:target 'app.storage.s3/head-object + :return (p/rejected (RuntimeException. "boom"))}] + ;; p/await returns the rejection wrapped in an ExecutionException + (let [ex (try + (impl/exists-object? (fake-s3-backend) {:id (uuid/next)}) + nil + (catch Throwable cause cause))] + (t/is (some? ex)) + (t/is (= "boom" (ex-message (ex-cause ex))))) + ;; one initial attempt plus max-retries + (t/is (= 4 (:call-count @mock)))))