mirror of
https://github.com/penpot/penpot.git
synced 2026-09-11 13:20:03 -04:00
Compare commits
8
Commits
develop
...
issue-11644
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ca46f2ad7f | ||
|
|
213a130b2e | ||
|
|
8d77db959e | ||
|
|
6d526efa17 | ||
|
|
a7021aa44c | ||
|
|
0fa3c28caf | ||
|
|
edd5a891a3 | ||
|
|
c09f790ba9 |
No files matched your search
@@ -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
|
||||
|
||||
@@ -390,7 +390,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)
|
||||
@@ -429,9 +428,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)}
|
||||
@@ -564,9 +560,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,19 @@
|
||||
[: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")
|
||||
|
||||
(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 +372,55 @@
|
||||
: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 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 below, 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 (-> (db/exec-one! pool [sql:link-upload-session-chunk
|
||||
(:id object) session-id index])
|
||||
(db/get-update-count))]
|
||||
(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 +438,37 @@
|
||||
: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 it died mid-flight and the client
|
||||
;; must start a new 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 +488,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"
|
||||
|
||||
@@ -162,6 +162,7 @@
|
||||
(= bucket "profile") (process-objects! conn has-profile-refs? bucket objects)
|
||||
(= bucket "file-data") (process-objects! conn has-file-data-refs? bucket objects)
|
||||
(= bucket sto/tempfile-bucket) (process-objects! conn (constantly false) sto/tempfile-bucket objects)
|
||||
(= bucket sto/upload-session-bucket) (process-objects! conn (constantly false) sto/upload-session-bucket objects)
|
||||
(= bucket "organization") (process-objects! conn (constantly false) bucket objects)
|
||||
:else
|
||||
(ex/raise :type :internal
|
||||
|
||||
@@ -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,7 +158,8 @@
|
||||
|
||||
(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 {})]
|
||||
@@ -224,7 +225,8 @@
|
||||
|
||||
(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 {})]
|
||||
@@ -271,7 +273,8 @@
|
||||
;; 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 {})]
|
||||
|
||||
@@ -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,158 @@
|
||||
(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-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
|
||||
|
||||
Reference in new issue
Block a user