mirror of
https://github.com/penpot/penpot.git
synced 2026-09-19 02:16:14 +00:00
✨ Add indirection for upload-chunks storage via new table (#11651)
* ✨ Add upload_session_chunk indirection for chunked uploads Chunks now live in the upload_session_chunk table with non-deleting foreign keys to storage_object and upload_session, instead of tempfile objects with session metadata. Reads go through a JOIN, so chunk state never scans storage_object. Uploads validate the live session, reject duplicate indexes, and store objects in the new upload-session bucket without extra metadata. Assemble removes mappings and marks the session consumed; objects-gc procedurally purges consumed and stalled sessions, touching referenced objects first. Touched-gc and deleted-gc handle the new bucket, and upload-session-gc is removed. Closes #11644 AI-assisted-by: muse-spark-1.3-contributor * 🐛 Fix quota, give-up and coverage for session chunks Exclude consumed sessions from the sessions-per-profile quota so finished uploads free their slot at once. Remove chunk mappings before the gc-deleted give-up delete to respect the NO ACTION keys. Catch java.sql.SQLException for duplicate chunks. Cover the profile-owned session purge and the UNIQUE race backstop with tests. AI-assisted-by: muse-spark-1.3-contributor * 🐛 Align chunked upload tests with upload_session_chunk Drop the duplicate-index tests written against metadata-backed chunks; the UNIQUE mapping makes those cases unrepresentable and the new tests cover them. Rewrite the rejected-duplicate tests to expect :validation/:chunk-already-exists and assert against the upload_session_chunk table, and scope the chunk-too-large "nothing stored" check to the mapping table. AI-assisted-by: muse-spark-1.3-contributor * ♻️ Use NO ACTION DEFERRABLE session FKs in single migration Fold the profile FK change into 0154 so the feature ships one migration. All three upload session FKs use ON DELETE NO ACTION DEFERRABLE: identical to RESTRICT in normal operation, but deferrable for tooling that relies on SET CONSTRAINTS ALL DEFERRED. Extend the RESTRICT test to the direct profile delete. AI-assisted-by: muse-spark-1.3-contributor * ♻️ Reserve chunk slot before writing blob in upload-chunk Make object_id nullable and insert the mapping with NULL inside the session-locking transaction, then write the blob outside it and link it with a conditional update. A failed write removes the mapping and reraises; a mid-flight death leaves a NULL row and the client starts a new session. AI-assisted-by: muse-spark-1.3-contributor * 🔥 Remove redundant session_id index on upload_session_chunk The UNIQUE(session_id, chunk_index) btree already serves session_id-only lookups and the session FK check through its leftmost column, so the standalone index only taxed the per-chunk INSERT path. Verified with EXPLAIN on an equivalent table shape. AI-assisted-by: muse-spark-1.3-contributor * ⚡ Merge chunk touch and delete into single RETURNING query Replace the SELECT-then-DELETE round-trip in delete-upload-sessions! with DELETE ... RETURNING object_id, touching each returned object. Same semantics, one less query per purged session. Follows the RETURNING pattern already used in file-gc. AI-assisted-by: muse-spark-1.3-contributor * ♻️ Let objects-gc own chunk mapping deletion Assemble-chunks now only marks the session as consumed; the chunk mappings stay until objects-gc purges them (touching the chunk objects first), leaving a single procedural deletion path for consumed, stalled and profile-purge sessions. AI-assisted-by: muse-spark-1.3-contributor * 🐛 Fix font-deletion GC expectations for chunk objects Update final storage-gc-touched counts to include the two chunk objects touched by objects-gc when purging consumed upload sessions (8/5/5 instead of 6/3/3). AI-assisted-by: muse-spark-1.3-contributor * 🐛 Release chunk reservation when the link UPDATE fails Review feedback on #11651: the link UPDATE in upload-chunk could leave a NULL reservation behind, blocking retries of the same index with :chunk-already-exists. Remove the reservation when the link fails so the client can retry in the same session; the orphaned blob stays touched for touched-gc. Also realign the process-bucket! cond branches in gc-touched. Tests: chunked-upload-link-failure-releases-slot and chunked-upload-null-reservation-blocks-retry. AI-assisted-by: muse-spark-1.3-contributor
This commit is contained in:
Andrey Antukh
GitHub
parent
b660ea9d53
commit
ad7e035b63
@ -86,7 +86,8 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
|
||||
| `file-thumbnail` | File grid thumbnails in `file_thumbnail.media_id`. | Yes | Authentication required | Reference scan. |
|
||||
| `profile` | User and team profile photos. References: `profile.photo_id` and `team.photo_id`. | Yes | Authentication required | Reference scan. |
|
||||
| `organization` | Organization logos uploaded by the Nitrate management API. | Yes | Public | No reference scan. A touched object is deleted. |
|
||||
| `tempfile` | Export files, chunked-upload chunks, and temporary font downloads. | No | Authentication required | No reference scan. A touched object uses a two-hour deletion delay. |
|
||||
| `tempfile` | Export files and temporary font downloads. | No | Authentication required | No reference scan. A touched object uses a two-hour deletion delay. |
|
||||
| `upload-session` | Chunked-upload chunks. References: `upload_session_chunk.object_id` and `upload_session_chunk.session_id` (both NO ACTION DEFERRABLE: restrict semantics, procedural deletion). | No | Authentication required | No reference scan. A touched object is deleted after the delay; `gc-deleted` removes mappings before rows. |
|
||||
| `file-data` | Encoded file data when `file-data-backend` is `storage`. Reference metadata has `storage-ref-id`, `file-id`, and the `file_data` row ID. | Yes | Authentication required | Reference scan. |
|
||||
| `file-data-fragment` | Compatibility value for file-data fragments. The current backend has no dedicated producer for this bucket. | No current write semantics | Public | No touched-object collector case. |
|
||||
| `file-change` | Compatibility value for file changes. Current snapshots store data in `file_data`, not this bucket. | No current write semantics | Authentication required | No touched-object collector case. |
|
||||
@ -95,7 +96,7 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
|
||||
- `file-media-object` is the default bucket for old rows without bucket metadata.
|
||||
- Do not assign a new bucket without adding its access and cleanup behavior.
|
||||
- The touched-object collector raises an internal error for an unknown bucket.
|
||||
- It supports `file-media-object`, `team-font-variant`, `file-object-thumbnail`, `file-thumbnail`, `profile`, `file-data`, `tempfile`, and `organization`.
|
||||
- It supports `file-media-object`, `team-font-variant`, `file-object-thumbnail`, `file-thumbnail`, `profile`, `file-data`, `tempfile`, `upload-session`, and `organization`.
|
||||
- It does not support `file-data-fragment` or `file-change`.
|
||||
|
||||
## Access Rules
|
||||
|
||||
@ -395,7 +395,6 @@
|
||||
:offload-file-data (ig/ref :app.tasks.offload-file-data/handler)
|
||||
:tasks-gc (ig/ref :app.tasks.tasks-gc/handler)
|
||||
:telemetry (ig/ref :app.tasks.telemetry/handler)
|
||||
: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)
|
||||
@ -434,9 +433,6 @@
|
||||
:app.tasks.tasks-gc/handler
|
||||
{::db/pool (ig/ref ::db/pool)}
|
||||
|
||||
:app.tasks.upload-session-gc/handler
|
||||
{::db/pool (ig/ref ::db/pool)}
|
||||
|
||||
:app.tasks.objects-gc/handler
|
||||
{::db/pool (ig/ref ::db/pool)
|
||||
::sto/storage (ig/ref ::sto/storage)}
|
||||
@ -569,9 +565,6 @@
|
||||
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
|
||||
:task :tasks-gc}
|
||||
|
||||
{:cron #penpot/cron "0 0 0 * * ?" ;; daily
|
||||
:task :upload-session-gc}
|
||||
|
||||
{:cron #penpot/cron "0 0 2 * * ?" ;; daily
|
||||
:task :file-gc-scheduler}
|
||||
|
||||
|
||||
@ -502,7 +502,10 @@
|
||||
: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")}])
|
||||
:fn (mg/resource "app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql")}
|
||||
|
||||
{:name "0154-add-upload-session-chunk-table"
|
||||
:fn (mg/resource "app/migrations/sql/0154-add-upload-session-chunk-table.sql")}])
|
||||
|
||||
(defn apply-migrations!
|
||||
[pool name migrations]
|
||||
|
||||
@ -0,0 +1,61 @@
|
||||
--- Add the upload_session_chunk table, a deleted_at marker to upload_session,
|
||||
--- and make the upload_session.profile_id foreign key non-deleting.
|
||||
|
||||
--- Each row maps one chunk of a chunked-upload session to the storage_object
|
||||
--- row that holds its bytes. Both foreign keys are ON DELETE NO ACTION
|
||||
--- DEFERRABLE on purpose: neither the session nor the storage object can be
|
||||
--- removed while a mapping row exists. Only objects-gc removes mappings
|
||||
--- (for consumed, stalled and profile-purge sessions), always before the
|
||||
--- session row, touching the chunk objects so storage GC reclaims them.
|
||||
--- NO ACTION is identical to RESTRICT in normal
|
||||
--- (immediate) operation; only the deferrability differs, which tooling
|
||||
--- such as the backend test fixture relies on
|
||||
--- (SET CONSTRAINTS ALL DEFERRED).
|
||||
---
|
||||
--- object_id is nullable: the mapping row is inserted first (reserving the
|
||||
--- slot under the UNIQUE(session_id, chunk_index) constraint inside a
|
||||
--- transaction that locks the session), and object_id is set once the blob
|
||||
--- has been written outside the transaction. A mapping with NULL object_id
|
||||
--- and no in-flight upload behind it means that upload died mid-flight; the
|
||||
--- client then starts a new session (sessions are ephemeral).
|
||||
|
||||
CREATE TABLE upload_session_chunk (
|
||||
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
|
||||
created_at timestamptz NOT NULL DEFAULT now(),
|
||||
|
||||
session_id uuid NOT NULL REFERENCES upload_session(id) ON DELETE NO ACTION DEFERRABLE,
|
||||
object_id uuid NULL REFERENCES storage_object(id) ON DELETE NO ACTION DEFERRABLE,
|
||||
chunk_index integer NOT NULL,
|
||||
|
||||
UNIQUE (session_id, chunk_index)
|
||||
);
|
||||
|
||||
--- No standalone index on session_id: the UNIQUE(session_id, chunk_index)
|
||||
--- btree already serves session_id-only lookups and the session FK check
|
||||
--- via its leftmost column.
|
||||
|
||||
CREATE INDEX upload_session_chunk__object_id__idx
|
||||
ON upload_session_chunk(object_id);
|
||||
|
||||
--- The deleted_at column marks a session as consumed: assemble-chunks sets it
|
||||
--- instead of deleting the row, and objects-gc physically purges consumed
|
||||
--- sessions (as well as stalled ones that were never assembled).
|
||||
|
||||
ALTER TABLE upload_session
|
||||
ADD COLUMN deleted_at timestamptz NULL DEFAULT NULL;
|
||||
|
||||
CREATE INDEX upload_session__deleted_at_created_at__idx
|
||||
ON upload_session(deleted_at, created_at);
|
||||
|
||||
--- The profile foreign key moves from CASCADE to NO ACTION DEFERRABLE:
|
||||
--- sessions must be purged procedurally (objects-gc drains the sessions of
|
||||
--- profiles pending purge before the profile row is deleted), so a profile
|
||||
--- can no longer disappear with live chunk mappings behind it.
|
||||
|
||||
ALTER TABLE upload_session
|
||||
DROP CONSTRAINT upload_session_profile_id_fkey;
|
||||
|
||||
ALTER TABLE upload_session
|
||||
ADD CONSTRAINT upload_session_profile_id_fkey
|
||||
FOREIGN KEY (profile_id) REFERENCES profile(id) ON DELETE NO ACTION DEFERRABLE;
|
||||
@ -339,7 +339,6 @@
|
||||
|
||||
;; --- Chunked Upload: Upload a single chunk
|
||||
|
||||
(declare ^:private get-upload-chunk)
|
||||
(declare ^:private check-upload-chunk-slot)
|
||||
|
||||
(def ^:private schema:upload-chunk
|
||||
@ -353,11 +352,27 @@
|
||||
[:session-id ::sm/uuid]
|
||||
[:index ::sm/int]])
|
||||
|
||||
(def ^:private sql:link-upload-session-chunk
|
||||
"UPDATE upload_session_chunk
|
||||
SET object_id = ?
|
||||
WHERE session_id = ?
|
||||
AND chunk_index = ?
|
||||
AND object_id IS NULL")
|
||||
|
||||
(defn- link-upload-session-chunk!
|
||||
"Links a stored blob to its reserved (session, index) slot. Returns the
|
||||
number of updated mappings (0 when the mapping vanished concurrently:
|
||||
the session was consumed or purged after the reserve)."
|
||||
[pool object-id session-id index]
|
||||
(-> (db/exec-one! pool [sql:link-upload-session-chunk object-id session-id index])
|
||||
(db/get-update-count)))
|
||||
|
||||
(sv/defmethod ::upload-chunk
|
||||
{::doc/added "2.17"
|
||||
::sm/params schema:upload-chunk
|
||||
::sm/result schema:upload-chunk-result}
|
||||
[cfg {:keys [::rpc/profile-id session-id index content]}]
|
||||
[{:keys [::db/pool] :as cfg}
|
||||
{:keys [::rpc/profile-id session-id index content] :as _params}]
|
||||
(let [session (db/tx-run! cfg check-upload-chunk-slot session-id profile-id index content)]
|
||||
(l/trc :hint "upload-chunk"
|
||||
:session-id session-id
|
||||
@ -365,23 +380,62 @@
|
||||
:size (:size content)
|
||||
:path (:path content))
|
||||
|
||||
;; NOTE: the blob is written outside any transaction on purpose (see
|
||||
;; mem:backend/storage): a failed write must never mingle with the
|
||||
;; mapping transaction. If the write or the link below fails, the
|
||||
;; reserved mapping is removed and the error propagates, so the client
|
||||
;; retries the index in the same session. If the process dies between
|
||||
;; the reserve and the link, a NULL mapping is left behind and the
|
||||
;; client starts a new session (sessions are ephemeral).
|
||||
(let [storage (sto/resolve cfg)
|
||||
data (sto/content (:path content))]
|
||||
(sto/put-object! storage
|
||||
{::sto/content data
|
||||
::sto/deduplicate? false
|
||||
::sto/touch true
|
||||
:content-type (:mtype content)
|
||||
:bucket sto/tempfile-bucket
|
||||
:upload-id (str session-id)
|
||||
:chunk-index index}))
|
||||
data (sto/content (:path content))
|
||||
object (try
|
||||
(sto/put-object! storage
|
||||
{::sto/content data
|
||||
::sto/deduplicate? false
|
||||
::sto/touched-at (ct/in-future {:hours 1})
|
||||
:content-type (:mtype content)
|
||||
:bucket sto/upload-session-bucket})
|
||||
(catch Throwable cause
|
||||
(db/delete! pool :upload-session-chunk
|
||||
{:session-id session-id :chunk-index index})
|
||||
(throw cause)))
|
||||
linked (try
|
||||
(link-upload-session-chunk! pool (:id object) session-id index)
|
||||
(catch Throwable cause
|
||||
;; The blob was stored but the link failed: drop the
|
||||
;; reservation so the client can retry the index in
|
||||
;; the same session; the orphaned object stays
|
||||
;; touched so touched-gc reclaims it.
|
||||
(db/delete! pool :upload-session-chunk
|
||||
{:session-id session-id :chunk-index index})
|
||||
(throw cause)))]
|
||||
(when (zero? linked)
|
||||
;; The mapping vanished concurrently (session consumed or purged
|
||||
;; after the reserve); the orphaned object stays touched so
|
||||
;; touched-gc reclaims it.
|
||||
(ex/raise :type :not-found
|
||||
:code :object-not-found
|
||||
:hint "upload session no longer available"
|
||||
:session-id session-id))))
|
||||
|
||||
{:session-id session-id
|
||||
:index index}))
|
||||
{:session-id session-id
|
||||
:index index})
|
||||
|
||||
(defn- check-upload-chunk-slot
|
||||
"Reserves the (session, index) slot: locks the session row, runs all
|
||||
validations and inserts the mapping with a NULL object_id, all in one
|
||||
transaction. Concurrent uploads of the same session serialize on the
|
||||
session lock, so the UNIQUE(session_id, chunk_index) constraint can
|
||||
never fire."
|
||||
[{:keys [::db/conn]} session-id profile-id index content]
|
||||
(let [session (db/get conn :upload-session {:id session-id :profile-id profile-id} {::db/for-update true})]
|
||||
(when (:deleted-at session)
|
||||
(ex/raise :type :not-found
|
||||
:code :object-not-found
|
||||
:hint "upload session already consumed"
|
||||
:session-id session-id))
|
||||
|
||||
(when (or (neg? index) (>= index (:total-chunks session)))
|
||||
(ex/raise :type :validation
|
||||
:code :invalid-chunk-index
|
||||
@ -399,40 +453,39 @@
|
||||
:size (:size content)
|
||||
:max-size (cf/get :upload-max-chunk-size)))
|
||||
|
||||
(when (get-upload-chunk conn session-id index)
|
||||
;; NOTE: a mapping with NULL object_id also counts as occupied: either
|
||||
;; its upload is still in flight, or its process died mid-flight and
|
||||
;; the client must start a new session. Known failures (write or link)
|
||||
;; remove the reservation above, so they stay retryable in the same
|
||||
;; session.
|
||||
(when (db/get* conn :upload-session-chunk {:session-id session-id :chunk-index index})
|
||||
(ex/raise :type :validation
|
||||
:code :duplicate-chunk-index
|
||||
:hint "chunk index already uploaded for this session"
|
||||
:code :chunk-already-exists
|
||||
:hint "chunk already uploaded for this session and index"
|
||||
:session-id session-id
|
||||
:index index))
|
||||
|
||||
(db/insert! conn :upload-session-chunk
|
||||
{:session-id session-id
|
||||
:object-id nil
|
||||
:chunk-index index})
|
||||
|
||||
session))
|
||||
|
||||
;; --- Chunked Upload: shared helpers
|
||||
|
||||
(def ^:private sql:get-upload-chunks
|
||||
"SELECT id, size, (metadata->>'~:chunk-index')::integer AS chunk_index
|
||||
FROM storage_object
|
||||
WHERE (metadata->>'~:upload-id') = ?::text
|
||||
AND deleted_at IS NULL
|
||||
AND status = 'valid'
|
||||
ORDER BY (metadata->>'~:chunk-index')::integer ASC")
|
||||
(def ^:private sql:get-upload-session-chunks
|
||||
"SELECT so.id, so.size
|
||||
FROM upload_session_chunk AS usc
|
||||
JOIN storage_object AS so ON (so.id = usc.object_id)
|
||||
WHERE usc.session_id = ?
|
||||
AND so.deleted_at IS NULL
|
||||
AND so.status = 'valid'
|
||||
ORDER BY usc.chunk_index ASC")
|
||||
|
||||
(defn- get-upload-chunks
|
||||
[conn session-id]
|
||||
(db/exec! conn [sql:get-upload-chunks (str session-id)]))
|
||||
|
||||
(def ^:private sql:get-upload-chunk
|
||||
"SELECT id
|
||||
FROM storage_object
|
||||
WHERE (metadata->>'~:upload-id') = ?::text
|
||||
AND (metadata->>'~:chunk-index')::integer = ?
|
||||
AND deleted_at IS NULL
|
||||
LIMIT 1")
|
||||
|
||||
(defn- get-upload-chunk
|
||||
[conn session-id index]
|
||||
(db/exec-one! conn [sql:get-upload-chunk (str session-id) index]))
|
||||
(db/exec! conn [sql:get-upload-session-chunks session-id]))
|
||||
|
||||
(defn- concat-chunks
|
||||
"Reads all chunk storage objects in order and writes them to a single
|
||||
@ -452,34 +505,47 @@
|
||||
conforming to `media.v/schema:upload` with `:filename`, `:path` and
|
||||
`:size`.
|
||||
|
||||
Raises a :validation/:missing-chunks error when the stored chunk
|
||||
indices do not form exactly the `0..total-chunks` range recorded in
|
||||
the session row (wrong count, gaps or duplicates).
|
||||
Raises :not-found when the session does not belong to `profile-id`.
|
||||
Deletes the session row from `upload_session` on success."
|
||||
Raises a :validation/:missing-chunks error when the number of stored
|
||||
chunks does not match `:total-chunks` recorded in the session row.
|
||||
Raises :not-found when the session does not belong to `profile-id` or
|
||||
was already consumed. Marks the session row as consumed (`deleted_at`);
|
||||
the chunk mappings stay until the objects-gc task purges them (touching
|
||||
the chunk objects so storage GC reclaims them), and the session row is
|
||||
purged afterwards."
|
||||
[{:keys [::db/conn] :as cfg} profile-id session-id]
|
||||
(let [session (db/get conn :upload-session {:id session-id :profile-id profile-id})
|
||||
chunks (get-upload-chunks conn session-id)
|
||||
indices (sort (map :chunk-index chunks))]
|
||||
(let [session (db/get conn :upload-session {:id session-id :profile-id profile-id})]
|
||||
(when (:deleted-at session)
|
||||
(ex/raise :type :not-found
|
||||
:code :object-not-found
|
||||
:hint "upload session already consumed"
|
||||
:session-id session-id))
|
||||
|
||||
(when (or (not= (count chunks) (:total-chunks session))
|
||||
(not= indices (range (:total-chunks session))))
|
||||
(ex/raise :type :validation
|
||||
:code :missing-chunks
|
||||
:hint "stored chunks do not match expected total"
|
||||
:session-id session-id
|
||||
:expected (:total-chunks session)
|
||||
:found (count chunks)))
|
||||
(let [chunks (get-upload-chunks conn session-id)]
|
||||
|
||||
(let [storage (sto/resolve cfg ::db/reuse-conn true)
|
||||
path (concat-chunks storage chunks)
|
||||
size (reduce #(+ %1 (:size %2)) 0 chunks)]
|
||||
(when (not= (count chunks) (:total-chunks session))
|
||||
(ex/raise :type :validation
|
||||
:code :missing-chunks
|
||||
:hint "number of stored chunks does not match expected total"
|
||||
:session-id session-id
|
||||
:expected (:total-chunks session)
|
||||
:found (count chunks)))
|
||||
|
||||
(db/delete! conn :upload-session {:id session-id})
|
||||
(let [storage (sto/resolve cfg ::db/reuse-conn true)
|
||||
path (concat-chunks storage chunks)
|
||||
size (reduce #(+ %1 (:size %2)) 0 chunks)]
|
||||
|
||||
{:filename "upload"
|
||||
:path path
|
||||
:size size})))
|
||||
;; NOTE: the session row is only marked (deleted_at) here; the
|
||||
;; chunk mappings stay until the objects-gc task removes them
|
||||
;; (before the session row, as the NO ACTION foreign keys
|
||||
;; require) while touching the chunk objects.
|
||||
(db/update! conn :upload-session
|
||||
{:deleted-at (ct/now)}
|
||||
{:id session-id}
|
||||
{::db/return-keys false})
|
||||
|
||||
{:filename "upload"
|
||||
:path path
|
||||
:size size}))))
|
||||
|
||||
;; --- Chunked Upload: Assemble all chunks into a final media object
|
||||
|
||||
|
||||
@ -535,7 +535,8 @@
|
||||
(def ^:private sql:get-upload-sessions-per-profile
|
||||
"SELECT count(*) AS total
|
||||
FROM upload_session
|
||||
WHERE profile_id = ?")
|
||||
WHERE profile_id = ?
|
||||
AND deleted_at IS NULL")
|
||||
|
||||
(defmethod check-quote ::upload-sessions-per-profile
|
||||
[{:keys [::profile-id ::target] :as quote}]
|
||||
|
||||
@ -42,6 +42,10 @@
|
||||
"Bucket name for temporary file uploads (10-minute expiry)."
|
||||
"tempfile")
|
||||
|
||||
(def upload-session-bucket
|
||||
"Bucket name for chunked-upload chunks."
|
||||
"upload-session")
|
||||
|
||||
(def valid-buckets
|
||||
#{"file-media-object"
|
||||
"team-font-variant"
|
||||
@ -50,6 +54,7 @@
|
||||
"profile"
|
||||
"organization"
|
||||
tempfile-bucket
|
||||
upload-session-bucket
|
||||
"file-data"
|
||||
"file-data-fragment"
|
||||
"file-change"})
|
||||
@ -211,7 +216,8 @@
|
||||
(if-some [hit (when (and (::deduplicate? params)
|
||||
(:hash mdata)
|
||||
(:bucket mdata)
|
||||
(not= tempfile-bucket (:bucket mdata)))
|
||||
(not= tempfile-bucket (:bucket mdata))
|
||||
(not= upload-session-bucket (:bucket mdata)))
|
||||
(get-database-object-by-hash pool backend
|
||||
(:bucket mdata)
|
||||
(:hash mdata)))]
|
||||
|
||||
@ -57,6 +57,18 @@
|
||||
(-> (db/exec-one! conn [sql:delete-sobjects ids])
|
||||
(db/get-update-count))))
|
||||
|
||||
(def ^:private sql:delete-upload-session-chunks
|
||||
"DELETE FROM upload_session_chunk
|
||||
WHERE object_id = ANY(?::uuid[])")
|
||||
|
||||
(defn- delete-upload-session-chunks!
|
||||
"Remove the chunk mappings for the given storage object ids. This must run
|
||||
before the storage_object rows are deleted: the upload_session_chunk
|
||||
foreign keys are ON DELETE NO ACTION."
|
||||
[conn ids]
|
||||
(let [ids (db/create-array conn "uuid" ids)]
|
||||
(db/exec-one! conn [sql:delete-upload-session-chunks ids])))
|
||||
|
||||
(def ^:private sql:increment-attempts-and-defer
|
||||
"UPDATE storage_object
|
||||
SET deletion_attempts = deletion_attempts + 1,
|
||||
@ -105,10 +117,21 @@
|
||||
:backend (name backend-id)))
|
||||
|
||||
(when (seq ok-ids)
|
||||
;; NOTE: the chunk mappings must be removed before the
|
||||
;; storage_object rows (NO ACTION foreign keys). It only affects
|
||||
;; objects of the upload-session bucket; for any other bucket the
|
||||
;; delete matches no rows.
|
||||
(delete-upload-session-chunks! conn ok-ids)
|
||||
(delete-sobjects! conn ok-ids))
|
||||
|
||||
(when (seq fail-ids)
|
||||
(increment-attempts-and-defer! conn fail-ids)
|
||||
;; NOTE: same NO ACTION ordering as above: the give-up DELETE below
|
||||
;; removes storage_object rows, so chunk mappings must go first.
|
||||
;; Deferred objects keep their rows; only the mapping of a
|
||||
;; permanently given-up object disappears early, and that object is
|
||||
;; already deleted-marked.
|
||||
(delete-upload-session-chunks! conn fail-ids)
|
||||
(let [given-up (delete-give-up! conn fail-ids)]
|
||||
(when (pos? (db/get-update-count given-up))
|
||||
(l/wrn :hint "giving up on orphan blob after max attempts"
|
||||
|
||||
@ -155,14 +155,15 @@
|
||||
(defn- process-bucket!
|
||||
[conn bucket objects]
|
||||
(cond
|
||||
(= bucket "file-media-object") (process-objects! conn has-file-media-object-refs? bucket objects)
|
||||
(= bucket "team-font-variant") (process-objects! conn has-team-font-variant-refs? bucket objects)
|
||||
(= bucket "file-object-thumbnail") (process-objects! conn has-file-object-thumbnails-refs? bucket objects)
|
||||
(= bucket "file-thumbnail") (process-objects! conn has-file-thumbnails-refs? bucket objects)
|
||||
(= bucket "profile") (process-objects! conn has-profile-refs? bucket objects)
|
||||
(= bucket "file-data") (process-objects! conn has-file-data-refs? bucket objects)
|
||||
(= bucket sto/tempfile-bucket) (process-objects! conn (constantly false) sto/tempfile-bucket objects)
|
||||
(= bucket "organization") (process-objects! conn (constantly false) bucket objects)
|
||||
(= bucket "file-media-object") (process-objects! conn has-file-media-object-refs? bucket objects)
|
||||
(= bucket "team-font-variant") (process-objects! conn has-team-font-variant-refs? bucket objects)
|
||||
(= bucket "file-object-thumbnail") (process-objects! conn has-file-object-thumbnails-refs? bucket objects)
|
||||
(= bucket "file-thumbnail") (process-objects! conn has-file-thumbnails-refs? bucket objects)
|
||||
(= bucket "profile") (process-objects! conn has-profile-refs? bucket objects)
|
||||
(= bucket "file-data") (process-objects! conn has-file-data-refs? bucket objects)
|
||||
(= bucket sto/tempfile-bucket) (process-objects! conn (constantly false) sto/tempfile-bucket objects)
|
||||
(= bucket sto/upload-session-bucket) (process-objects! conn (constantly false) sto/upload-session-bucket objects)
|
||||
(= bucket "organization") (process-objects! conn (constantly false) bucket objects)
|
||||
:else
|
||||
(ex/raise :type :internal
|
||||
:code :unexpected-unknown-reference
|
||||
|
||||
@ -16,6 +16,51 @@
|
||||
[app.tasks.delete-object :as dobj]
|
||||
[integrant.core :as ig]))
|
||||
|
||||
(def ^:private sql:get-upload-sessions
|
||||
"SELECT us.id
|
||||
FROM upload_session AS us
|
||||
WHERE (us.deleted_at IS NOT NULL
|
||||
AND us.deleted_at <= ?)
|
||||
OR (us.deleted_at IS NULL
|
||||
AND us.created_at <= ?)
|
||||
OR EXISTS (SELECT 1
|
||||
FROM profile AS p
|
||||
WHERE p.id = us.profile_id
|
||||
AND p.deleted_at IS NOT NULL
|
||||
AND p.deleted_at <= ?)
|
||||
ORDER BY us.created_at ASC
|
||||
LIMIT ?
|
||||
FOR UPDATE OF us
|
||||
SKIP LOCKED")
|
||||
|
||||
(def ^:private sql:delete-session-chunks
|
||||
"DELETE FROM upload_session_chunk
|
||||
WHERE session_id = ?
|
||||
RETURNING object_id")
|
||||
|
||||
(defn- delete-upload-sessions!
|
||||
"Purges consumed upload sessions (marked by assemble-chunks), stalled
|
||||
sessions (never assembled within max-age) and sessions owned by profiles
|
||||
pending purge. Referenced storage objects are touched so the storage GC
|
||||
reclaims them with its usual delay; chunk mappings are removed before the
|
||||
session row (NO ACTION foreign keys)."
|
||||
[{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
|
||||
(let [stalled-threshold (ct/minus timestamp {:hours 1})]
|
||||
(->> (db/plan conn [sql:get-upload-sessions timestamp stalled-threshold timestamp chunk-size]
|
||||
{:fetch-size 5})
|
||||
(reduce (fn [total {:keys [id]}]
|
||||
(l/trc :obj "upload-session" :id (str id))
|
||||
|
||||
;; Remove the chunk mappings, marking as touched all
|
||||
;; related storage objects in a single round-trip.
|
||||
(doseq [{:keys [object-id]} (db/exec! conn [sql:delete-session-chunks id])]
|
||||
(some->> object-id (sto/touch-object! storage)))
|
||||
|
||||
(let [affected (-> (db/delete! conn :upload-session {:id id})
|
||||
(db/get-update-count))]
|
||||
(+ total affected)))
|
||||
0))))
|
||||
|
||||
(def ^:private sql:get-profiles
|
||||
"SELECT id, photo_id FROM profile
|
||||
WHERE deleted_at IS NOT NULL
|
||||
@ -292,7 +337,11 @@
|
||||
0)))
|
||||
|
||||
(def ^:private deletion-proc-vars
|
||||
[#'delete-profiles!
|
||||
;; NOTE: upload sessions go first: deleting a profile cascades to its
|
||||
;; sessions, which would hit the upload_session_chunk NO ACTION foreign key
|
||||
;; while mappings still exist.
|
||||
[#'delete-upload-sessions!
|
||||
#'delete-profiles!
|
||||
#'delete-file-media-objects!
|
||||
#'delete-file-object-thumbnails!
|
||||
#'delete-file-thumbnails!
|
||||
|
||||
@ -1,41 +0,0 @@
|
||||
;; 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 SUBSIDIARY SL
|
||||
|
||||
(ns app.tasks.upload-session-gc
|
||||
"A maintenance task that deletes stalled (incomplete) upload sessions.
|
||||
|
||||
An upload session is considered stalled when it was created more than
|
||||
`max-age` ago without being completed (i.e. the session row still
|
||||
exists because `assemble-chunks` was never called to clean it up).
|
||||
The default max-age is 1 hour."
|
||||
(:require
|
||||
[app.common.logging :as l]
|
||||
[app.common.time :as ct]
|
||||
[app.db :as db]
|
||||
[integrant.core :as ig]))
|
||||
|
||||
(def ^:private sql:delete-stalled-sessions
|
||||
"DELETE FROM upload_session
|
||||
WHERE created_at < ?::timestamptz")
|
||||
|
||||
(defmethod ig/assert-key ::handler
|
||||
[_ params]
|
||||
(assert (db/pool? (::db/pool params)) "expected a valid database pool"))
|
||||
|
||||
(defmethod ig/expand-key ::handler
|
||||
[k v]
|
||||
{k (merge {::max-age (ct/duration {:hours 1})} v)})
|
||||
|
||||
(defmethod ig/init-key ::handler
|
||||
[_ {:keys [::max-age] :as cfg}]
|
||||
(fn [_]
|
||||
(db/tx-run! cfg
|
||||
(fn [{:keys [::db/conn]}]
|
||||
(let [threshold (ct/minus (ct/now) max-age)
|
||||
result (-> (db/exec-one! conn [sql:delete-stalled-sessions threshold])
|
||||
(db/get-update-count))]
|
||||
(l/debug :hint "task finished" :deleted result)
|
||||
{:deleted result})))))
|
||||
@ -158,12 +158,15 @@
|
||||
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 2 (:processed res)))))
|
||||
;; processed = 4: the 2 font variants plus the 2 consumed upload sessions
|
||||
(t/is (= 4 (:processed res)))))
|
||||
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
|
||||
(let [res (th/run-task! :storage-gc-touched {})]
|
||||
(t/is (= 0 (:freeze res)))
|
||||
(t/is (= 6 (:delete res)))))))
|
||||
;; deleted = 8: the 6 font objects plus the 2 chunk objects touched
|
||||
;; by objects-gc when purging the consumed sessions
|
||||
(t/is (= 8 (:delete res)))))))
|
||||
|
||||
(t/deftest font-deletion-2
|
||||
(let [prof (th/create-profile* 1 {:is-active true})
|
||||
@ -224,12 +227,15 @@
|
||||
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 1 (:processed res)))))
|
||||
;; processed = 3: the font plus the 2 consumed upload sessions
|
||||
(t/is (= 3 (:processed res)))))
|
||||
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
|
||||
(let [res (th/run-task! :storage-gc-touched {})]
|
||||
(t/is (= 0 (:freeze res)))
|
||||
(t/is (= 3 (:delete res)))))))
|
||||
;; deleted = 5: the 3 font objects plus the 2 chunk objects touched
|
||||
;; by objects-gc when purging the consumed sessions
|
||||
(t/is (= 5 (:delete res)))))))
|
||||
|
||||
(t/deftest font-deletion-3
|
||||
(let [prof (th/create-profile* 1 {:is-active true})
|
||||
@ -271,12 +277,15 @@
|
||||
;; objects-gc at days 8, then storage-gc-touched at days 8 + 3h
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 1 (:processed res)))))
|
||||
;; processed = 3: the font variant plus the 2 consumed upload sessions
|
||||
(t/is (= 3 (:processed res)))))
|
||||
|
||||
(binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
|
||||
(let [res (th/run-task! :storage-gc-touched {})]
|
||||
(t/is (= 0 (:freeze res)))
|
||||
(t/is (= 3 (:delete res)))))))
|
||||
;; deleted = 5: the 3 font objects plus the 2 chunk objects touched
|
||||
;; by objects-gc when purging the consumed sessions
|
||||
(t/is (= 5 (:delete res)))))))
|
||||
|
||||
(t/deftest input-sanitization-1
|
||||
(with-mocks [mock {:target 'app.rpc.quotes/check! :return nil}]
|
||||
|
||||
@ -7,6 +7,7 @@
|
||||
(ns backend-tests.rpc-media-test
|
||||
(:require
|
||||
[app.common.uuid :as uuid]
|
||||
[app.db :as db]
|
||||
[app.http.client :as http]
|
||||
[app.media :as media]
|
||||
[app.rpc :as-alias rpc]
|
||||
@ -532,7 +533,7 @@
|
||||
:index 0
|
||||
:content mfile})
|
||||
|
||||
;; First assemble succeeds; session row is deleted afterwards
|
||||
;; First assemble succeeds; session row is marked as consumed afterwards
|
||||
(let [out1 (th/command! {::th/type :assemble-file-media-object
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
@ -545,7 +546,7 @@
|
||||
(t/is (= media-id (:id (:result out1)))))
|
||||
|
||||
;; Second assemble with the same session-id must fail because the
|
||||
;; session row has been deleted after the first assembly
|
||||
;; session row has been marked as consumed after the first assembly
|
||||
(let [out2 (th/command! {::th/type :assemble-file-media-object
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
@ -681,45 +682,6 @@
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :missing-chunks (-> out :error ex-data :code))))))
|
||||
|
||||
(t/deftest chunked-upload-assemble-rejects-duplicate-indices
|
||||
;; assemble-chunks must validate the index SET, not just the count: a
|
||||
;; session declaring 2 chunks but storing [0,0] must fail instead of
|
||||
;; assembling a corrupt file. Chunks are written at the storage level
|
||||
;; because upload-chunk itself now rejects the second index.
|
||||
(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})
|
||||
session-id (create-session! prof 2)
|
||||
storage (:app.storage/storage th/*system*)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
chunks (split-file-into-chunks source-path 312043)
|
||||
put-chunk! (fn [idx]
|
||||
(let [mfile (make-chunk-mfile (first chunks) "image/jpeg")]
|
||||
(sto/put-object! storage
|
||||
{::sto/content (sto/content (:path mfile))
|
||||
::sto/deduplicate? false
|
||||
::sto/touch true
|
||||
:content-type "image/jpeg"
|
||||
:bucket sto/tempfile-bucket
|
||||
:upload-id (str session-id)
|
||||
:chunk-index idx})))]
|
||||
(put-chunk! 0)
|
||||
(put-chunk! 0)
|
||||
|
||||
(let [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 "dupe-indices"
|
||||
:mtype "image/jpeg"})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :missing-chunks (-> out :error ex-data :code))))))
|
||||
|
||||
(t/deftest chunked-upload-duplicate-then-assemble
|
||||
;; A rejected duplicate must leave the first chunk intact: upload 0,
|
||||
;; re-upload 0 (rejected), then assemble succeeds with the original size.
|
||||
@ -748,7 +710,8 @@
|
||||
:index 0
|
||||
:content (make-chunk-mfile (first chunks) mtype)})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :chunk-already-exists (-> out :error ex-data :code))))
|
||||
|
||||
(let [out (th/command! {::th/type :assemble-file-media-object
|
||||
::rpc/profile-id (:id prof)
|
||||
@ -791,7 +754,8 @@
|
||||
:index 0
|
||||
:content (make-chunk-mfile (nth chunks 0) mtype)})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :chunk-already-exists (-> out :error ex-data :code))))
|
||||
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
@ -800,11 +764,11 @@
|
||||
:content (make-chunk-mfile (nth chunks 1) mtype)})]
|
||||
(t/is (nil? (:error out))))
|
||||
|
||||
;; The live store holds exactly the two distinct indices: the
|
||||
;; The mapping table holds exactly the two distinct indices: the
|
||||
;; rejected duplicate stored nothing.
|
||||
(let [rows (th/db-exec! ["SELECT (metadata->>'~:chunk-index')::integer AS idx FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND deleted_at IS NULL ORDER BY idx"
|
||||
(str session-id)])]
|
||||
(t/is (= [0 1] (mapv :idx rows))))))
|
||||
(let [rows (th/db-exec! ["SELECT chunk_index FROM upload_session_chunk WHERE session_id = ? ORDER BY chunk_index"
|
||||
session-id])]
|
||||
(t/is (= [0 1] (mapv :chunk-index rows))))))
|
||||
|
||||
(t/deftest chunked-upload-session-not-found
|
||||
(let [prof (th/create-profile* 1)
|
||||
@ -843,6 +807,48 @@
|
||||
(t/is (= :max-quote-reached (-> out :error ex-data :code)))
|
||||
(t/is (= "upload-chunks-per-session" (-> out :error ex-data :target))))))
|
||||
|
||||
(t/deftest chunked-upload-consumed-session-frees-quota
|
||||
;; Consumed sessions must not count against the sessions-per-profile
|
||||
;; quota: with the limit set to 1, assembling a session frees the slot
|
||||
;; for a new one.
|
||||
(with-mocks [mock {:target 'app.config/get
|
||||
:return (th/config-get-mock
|
||||
{:quotes-upload-sessions-per-profile 1})}]
|
||||
(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})
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
session-id (create-session! prof 1)
|
||||
upload-out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (nil? (:error upload-out)))
|
||||
|
||||
(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 (nil? (:error assemble-out))))
|
||||
|
||||
;; the consumed session frees the quota slot
|
||||
(let [out (th/command! {::th/type :create-upload-session
|
||||
::rpc/profile-id (:id prof)
|
||||
:total-chunks 1})]
|
||||
(t/is (nil? (:error out)))
|
||||
(t/is (uuid? (:session-id (:result out))))))))
|
||||
|
||||
(t/deftest chunked-upload-invalid-total-chunks
|
||||
;; total-chunks must be at least 1; zero and negative values are rejected
|
||||
;; with a :validation error.
|
||||
@ -892,41 +898,6 @@
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :invalid-chunk-index (-> out :error ex-data :code))))))
|
||||
|
||||
(t/deftest chunked-upload-duplicate-index-rejected
|
||||
;; Uploading the same chunk index twice into one session must fail:
|
||||
;; the second call raises :validation / :duplicate-chunk-index and
|
||||
;; stores nothing, so one session+index keeps at most one object.
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
chunks (split-file-into-chunks source-path 312043)
|
||||
mtype "image/jpeg"
|
||||
mfile1 (make-chunk-mfile (first chunks) mtype)
|
||||
mfile2 (make-chunk-mfile (first chunks) mtype)]
|
||||
|
||||
;; First upload succeeds
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile1})]
|
||||
(t/is (nil? (:error out))))
|
||||
|
||||
;; Second upload of the same index must be rejected
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile2})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
|
||||
|
||||
;; Exactly one live object stored for that session/index
|
||||
(let [rows (th/db-exec! ["SELECT id FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND (metadata->>'~:chunk-index') = '0' AND deleted_at IS NULL"
|
||||
(str session-id)])]
|
||||
(t/is (= 1 (count rows))))))
|
||||
|
||||
(t/deftest chunked-upload-chunk-too-large
|
||||
;; Chunks larger than the configured cap must be rejected with
|
||||
;; :validation / :chunk-too-large before anything is stored, while a
|
||||
@ -951,9 +922,8 @@
|
||||
(t/is (= :chunk-too-large (-> out :error ex-data :code))))
|
||||
|
||||
;; Nothing stored for the rejected chunk
|
||||
(let [rows (th/db-exec! ["SELECT id FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND deleted_at IS NULL"
|
||||
(str session-id)])]
|
||||
(t/is (= 0 (count rows))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["SELECT count(*) FROM upload_session_chunk WHERE session_id = ?"
|
||||
session-id]))))
|
||||
|
||||
;; A chunk exactly at the cap still uploads fine
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
@ -984,6 +954,220 @@
|
||||
(t/is (= :restriction (-> out :error ex-data :type)))
|
||||
(t/is (= :max-quote-reached (-> out :error ex-data :code)))))))
|
||||
|
||||
;; --- upload_session_chunk mapping tests ---
|
||||
|
||||
(t/deftest chunked-upload-creates-chunk-mapping
|
||||
;; Uploading a chunk creates a row in upload_session_chunk pointing to the
|
||||
;; storage object, and the object itself carries no session metadata.
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
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)))
|
||||
|
||||
(let [row (th/db-exec-one! ["select session_id, object_id, chunk_index from upload_session_chunk where session_id = ?"
|
||||
session-id])]
|
||||
(t/is (= session-id (:session-id row)))
|
||||
(t/is (= 0 (:chunk-index row)))
|
||||
|
||||
(let [storage (:app.storage/storage th/*system*)
|
||||
obj (sto/get-object storage (:object-id row))]
|
||||
(t/is (sto/object? obj))
|
||||
(t/is (= "upload-session" (-> obj meta :bucket)))
|
||||
(t/is (nil? (-> obj meta :upload-id)))
|
||||
(t/is (nil? (-> obj meta :chunk-index)))))))
|
||||
|
||||
(t/deftest chunked-upload-duplicate-index-fails
|
||||
;; Re-uploading an already stored index fails with
|
||||
;; :validation/:chunk-already-exists and creates no new storage object.
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
out1 (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (nil? (:error out1)))
|
||||
|
||||
(let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
|
||||
out2 (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (some? (:error out2)))
|
||||
(t/is (= :validation (-> out2 :error ex-data :type)))
|
||||
(t/is (= :chunk-already-exists (-> out2 :error ex-data :code)))
|
||||
(t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
|
||||
|
||||
(t/deftest chunked-upload-null-reservation-blocks-retry
|
||||
;; A reserved slot with NULL object_id (an upload that died between the
|
||||
;; reserve and the link) counts as occupied: retrying the index in the
|
||||
;; same session fails with :validation/:chunk-already-exists and stores
|
||||
;; nothing, so the client must start a new session.
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}]
|
||||
(th/db-exec! ["insert into upload_session_chunk (session_id, chunk_index, object_id) values (?, ?, null)"
|
||||
session-id 0])
|
||||
(let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
|
||||
out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :validation (-> out :error ex-data :type)))
|
||||
(t/is (= :chunk-already-exists (-> out :error ex-data :code)))
|
||||
(t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
|
||||
|
||||
(t/deftest chunked-upload-link-failure-releases-slot
|
||||
;; When the link UPDATE fails after a successful blob write, the
|
||||
;; reservation is removed so the client can retry the index in the same
|
||||
;; session; the orphaned blob stays touched for touched-gc.
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
orig @#'app.rpc.commands.media/link-upload-session-chunk!
|
||||
failed? (atom false)]
|
||||
(with-mocks [mock {:target 'app.rpc.commands.media/link-upload-session-chunk!
|
||||
:return (fn [pool object-id session-id index]
|
||||
(if (compare-and-set! failed? false true)
|
||||
(throw (ex-info "link boom" {}))
|
||||
(orig pool object-id session-id index)))}]
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (some? (:error out))))
|
||||
;; the failed link left no reservation behind
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
;; retrying the same index in the same session succeeds
|
||||
(let [out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (nil? (:error out)))
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))))))
|
||||
|
||||
(t/deftest chunked-upload-to-consumed-session-fails
|
||||
;; Once assembled, the session is consumed: uploading another chunk fails
|
||||
;; with :not-found and the session row stays, marked with deleted_at.
|
||||
(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})
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
out1 (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (nil? (:error out1)))
|
||||
|
||||
(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 (nil? (:error assemble-out))))
|
||||
|
||||
;; chunk mappings stay until objects-gc purges them, session row
|
||||
;; stays marked as consumed
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
(t/is (some? (:deleted-at (th/db-exec-one! ["select deleted_at from upload_session where id = ?"
|
||||
session-id]))))
|
||||
|
||||
;; uploading to the consumed session fails without creating an object
|
||||
(let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
|
||||
out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
(t/is (some? (:error out)))
|
||||
(t/is (= :not-found (-> out :error ex-data :type)))
|
||||
(t/is (= :object-not-found (-> out :error ex-data :code)))
|
||||
(t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
|
||||
|
||||
(defn- sql-state-of
|
||||
"Runs thunk (a db statement) and returns the SQLState of the raised
|
||||
SQLException, or nil when no error is raised."
|
||||
[thunk]
|
||||
(try
|
||||
(thunk)
|
||||
nil
|
||||
(catch java.sql.SQLException cause
|
||||
(.getSQLState cause))))
|
||||
|
||||
(t/deftest upload-session-chunk-restrict-blocks-direct-deletes
|
||||
;; With a live mapping row, deleting the storage object or the session
|
||||
;; directly violates the RESTRICT foreign keys (SQLState 23503).
|
||||
(let [prof (th/create-profile* 1)
|
||||
session-id (create-session! prof 1)
|
||||
source-path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
mfile {:filename "sample.jpg"
|
||||
:path source-path
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
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)))
|
||||
|
||||
(let [object-id (:object-id (th/db-exec-one! ["select object_id from upload_session_chunk where session_id = ?"
|
||||
session-id]))]
|
||||
(t/is (= "23503" (sql-state-of #(th/db-exec! ["delete from storage_object where id = ?"
|
||||
object-id]))))
|
||||
(t/is (= "23503" (sql-state-of #(th/db-exec! ["delete from upload_session where id = ?"
|
||||
session-id]))))
|
||||
;; the profile cannot disappear either while its session is live
|
||||
;; (profile_id FK is NO ACTION DEFERRABLE; purge goes through
|
||||
;; objects-gc). The deletion_protection rule is disabled here so the
|
||||
;; statement reaches the FK check.
|
||||
(t/is (= "23503" (sql-state-of #(db/transact! th/*pool*
|
||||
(fn [conn]
|
||||
(db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
|
||||
(db/exec! conn ["delete from profile where id = ?"
|
||||
(:id prof)])))))))))
|
||||
|
||||
;; --- Clone File Media Object BOLA tests ---
|
||||
|
||||
(defn- create-storage-object!
|
||||
|
||||
@ -292,8 +292,9 @@
|
||||
{:id (:id result-2)})
|
||||
|
||||
;; run the objects gc task for permanent deletion
|
||||
;; (processed = 2: the consumed upload session plus the font variant)
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 1 (:processed res))))
|
||||
(t/is (= 2 (:processed res))))
|
||||
|
||||
;; revert touched state to all storage objects
|
||||
|
||||
@ -817,8 +818,8 @@
|
||||
|
||||
;; 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)])
|
||||
(th/db-exec! ["update storage_object set status = 'pending' where id in (select object_id from upload_session_chunk where session_id = ?)"
|
||||
session-id])
|
||||
|
||||
;; assembling fails because no chunk is visible anymore
|
||||
(let [assemble-out (th/command! {::th/type :assemble-file-media-object
|
||||
@ -830,6 +831,143 @@
|
||||
:mtype "image/jpeg"})]
|
||||
(t/is (some? (:error assemble-out))))))
|
||||
|
||||
(t/deftest upload-session-stalled-purge-lifecycle
|
||||
;; Full lifecycle of a stalled session: objects-gc purges the session and
|
||||
;; its mappings while touching the objects, touched-gc marks them deleted
|
||||
;; and deleted-gc removes rows and blobs.
|
||||
(let [prof (th/create-profile* 1)
|
||||
_ (th/create-project* 1 {:profile-id (:id prof)
|
||||
:team-id (:default-team-id prof)})
|
||||
_ (th/create-file* 1 {:profile-id (:id prof)
|
||||
:project-id (:default-project-id prof)
|
||||
:is-shared false})
|
||||
mfile {:filename "chunk"
|
||||
:path (th/tempfile "backend_tests/test_files/sample.jpg")
|
||||
:mtype "image/jpeg"
|
||||
:size 312043}
|
||||
session-id (-> (th/command! {::th/type :create-upload-session
|
||||
::rpc/profile-id (:id prof)
|
||||
:total-chunks 1})
|
||||
:result :session-id)
|
||||
out (th/command! {::th/type :upload-chunk
|
||||
::rpc/profile-id (:id prof)
|
||||
:session-id session-id
|
||||
:index 0
|
||||
:content mfile})]
|
||||
|
||||
(t/is (nil? (:error out)))
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
|
||||
;; backdate the session so it counts as stalled
|
||||
(th/db-exec! ["update upload_session set created_at = now() - interval '2 hours' where id = ?"
|
||||
session-id])
|
||||
|
||||
;; objects-gc purges session and mappings, touching the objects
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 1 (:processed res))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
|
||||
session-id]))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))
|
||||
|
||||
;; touched-gc marks the orphaned object as deleted
|
||||
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
|
||||
(th/run-task! :storage-gc-touched {}))]
|
||||
(t/is (= 0 (:freeze res)))
|
||||
(t/is (= 1 (:delete res))))
|
||||
|
||||
;; deleted-gc removes the row and the blob (clock past the mark time)
|
||||
(let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 4}))]
|
||||
(th/run-task! :storage-gc-deleted {}))]
|
||||
(t/is (= 1 (:deleted res))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object"]))))))
|
||||
|
||||
(t/deftest upload-session-consumed-purge
|
||||
;; An assembled session is marked as consumed and objects-gc purges it
|
||||
;; right away, without waiting for the stalled threshold.
|
||||
(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)))
|
||||
|
||||
(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 (nil? (:error assemble-out))))
|
||||
|
||||
;; mappings stay, session row stays marked as consumed; objects-gc
|
||||
;; purges both
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
(t/is (some? (:deleted-at (th/db-exec-one! ["select deleted_at from upload_session where id = ?"
|
||||
session-id]))))
|
||||
|
||||
;; objects-gc purges the consumed session immediately
|
||||
(let [res (th/run-task! :objects-gc {})]
|
||||
(t/is (= 1 (:processed res))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
|
||||
session-id]))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))))
|
||||
|
||||
(t/deftest upload-session-profile-purge
|
||||
;; Sessions owned by a profile pending purge are drained first, so the
|
||||
;; profile delete (which cascades to its sessions) never hits the chunk
|
||||
;; RESTRICT foreign keys.
|
||||
(let [prof (th/create-profile* 1)
|
||||
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)))
|
||||
|
||||
;; soft-delete the profile; the live session is neither consumed nor stalled
|
||||
(th/db-update! :profile {:deleted-at (ct/now)} {:id (:id prof)})
|
||||
|
||||
(th/run-task! :objects-gc {})
|
||||
|
||||
;; session and mappings are gone, profile row deletes cleanly
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
|
||||
session-id]))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
|
||||
session-id]))))
|
||||
(t/is (= 0 (:count (th/db-exec-one! ["select count(*) from profile where id = ?"
|
||||
(:id prof)]))))
|
||||
;; and the chunk object was touched for the storage GC
|
||||
(t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))))
|
||||
|
||||
(defn- fake-s3-backend
|
||||
[]
|
||||
{::sto/type :s3
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user