github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-08-27 23:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Add storage object status lifecycle and verified dedup (#11345)

Browse Source 
* ♻️ Simplify storage GC delays and add skip-delay task params

The touched GC no longer applies an extra deletion-delay when marking
storage objects as deleted. By the time a storage object is touched, its
referencing domain row has already passed its own deletion delay, and the
reference scan is the only safety check needed. Touched objects are now
marked with deleted_at = now, so the deleted GC removes them on the next
run.

For the tempfile bucket, upload chunks now set touched-at in the future
(1h, aligned with the upload-session-gc TTL) instead of relying on a
special-case deletion delay.

Task handlers now read their task props:
- storage-gc-touched accepts :skip-delay to process all touched objects
  immediately, bypassing the min-age threshold.
- objects-gc accepts :chunk-size and :skip-delay to process recently
  deleted rows without waiting for the deletion delay.

This allows running the deletion cascade immediately from the REPL via
run-task! with the skip-delay option.

AI-assisted-by: deepseek-v4-flash

* ✨ Add storage object status lifecycle, verified dedup, and deletion retry tracking

Storage object lifecycle hardening:

- Add status column ('valid' | 'pending') as write-ahead marker for
  object creation. put-object! inserts in 'pending' state, writes blob,
  then promotes to 'valid'. Failed writes remove the pending row.
- Add :storage-pending-gc task to reclaim orphaned pending rows (e.g.
  after crash between blob write and promotion).
- Verify blob existence on every dedup hit via exists-object? (fs stat /
  s3 headObject). Missing blobs mark the row as deleted and create fresh
  object.
- Add deletion_attempts column (migration 0154) to track physical blob
  deletion attempts. Restructure gc_deleted to use chunked processing
  with per-chunk transactions (short lock duration). Failed deletions
  are deferred to tomorrow (deleted_at = NOW() + 1 day) to prevent
  infinite loops. After 7 attempts, give up and accept orphan.
- Change del-objects-in-bulk contract to return #{fail-ids} for precise
  per-id tracking (fs and s3 backends updated).
- Use tmp/tempfile for fs atomic writes with cleanup queue registration
  (crashed-JVM temp files swept ~60min later). Document ATOMIC_MOVE
  POSIX-only assumption.
- Add linear backoff to s3 exists-object? retries (100ms/200ms/300ms).
- Wrap compensating delete in put-object! catch block to prevent
  masking original error when connection is aborted.
- Fix assert messages in pending_gc.clj and gc_deleted.clj (pool
  assertion said 'expected valid storage' instead of 'db pool').
- Add pending-objects-excluded-from-gc-deleted test. Use unique path in
  put-object-write-failure-leaves-no-row test to avoid collisions.

AI-assisted-by: qwen3.7-plus

* 🐛 Fix review comments on gc-deleted and storage

- Fix process-chunk! returning nil causing (+ acc nil) crash
- Add FOR UPDATE SKIP LOCKED to sql:get-deleted-chunk to prevent
  infinite loop when another worker holds locks
- Pass :cause to log messages in gc_deleted.clj and s3.clj
- Fix extra space in log hint string
- Remove unused ::blob-missing? reference from storage memory
- Rename test to match actual behavior (leaves pending row)
- Add test for gc-deleted giving up after max attempts

AI-assisted-by: qwen3.7-plus
This commit is contained in:
Andrey Antukh 2026-08-27 12:37:05 +02:00 committed by GitHub
parent a3feb4ef3b
commit 0e388442a1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 1064 additions and 228 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
.serena/memories/backend/storage.md
View File

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

22
backend/src/app/binfile/common.clj
View File

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

30
backend/src/app/binfile/v3.clj
View File

@ -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]

7
backend/src/app/features/fdata.clj
View File

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

5
backend/src/app/features/file_snapshots.clj
View File

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

9
backend/src/app/main.clj
View File

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

5
backend/src/app/migrations.clj
View File

@ -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]

24
backend/src/app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql Normal file
View File

@ -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;

46
backend/src/app/rpc/commands/files_thumbnails.clj
View File

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

3
backend/src/app/rpc/commands/media.clj
View File

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

177
backend/src/app/storage.clj
View File

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

62
backend/src/app/storage/fs.clj
View File

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

144
backend/src/app/storage/gc_deleted.clj
View File

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

21
backend/src/app/storage/gc_touched.clj
View File

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

13
backend/src/app/storage/impl.clj
View File

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

88
backend/src/app/storage/pending_gc.clj Normal file
View File

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

74
backend/src/app/storage/s3.clj
View File

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

7
backend/src/app/storage/tmp.clj
View File

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

9
backend/src/app/tasks/file_gc.clj
View File

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

10
backend/src/app/tasks/objects_gc.clj
View File

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

500
backend/test/backend_tests/storage_test.clj
View File

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