diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md index 4bfa51e914..6ab862a311 100644 --- a/.serena/memories/backend/storage.md +++ b/.serena/memories/backend/storage.md @@ -86,7 +86,7 @@ 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 and temporary font downloads. Legacy chunked-upload chunks (pre-`upload_session_chunk`) still expire through this bucket. | 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. | diff --git a/.serena/memories/prod-infra/core.md b/.serena/memories/prod-infra/core.md index 6c64ee4763..1ec5af0308 100644 --- a/.serena/memories/prod-infra/core.md +++ b/.serena/memories/prod-infra/core.md @@ -22,7 +22,7 @@ Postgres row locking is the only correctness primitive: `task` claims via `FOR U Two known race patterns survive multi-backend operation: -- **Cron dedup is best-effort.** The lock on `scheduled_task` is released when the task body finishes. If two backends' cron timers fire for the same scheduled instant with a gap larger than the task body's runtime, both execute it. Penpot's cron entries are idempotent (`session-gc`, `objects-gc` (incl. upload-session purge), `storage-gc-*`, `tasks-gc`, `file-gc-scheduler`); the exceptions are `:telemetry` (would double-report) and `:audit-log-archive` (depends on archive target idempotency). +- **Cron dedup is best-effort.** The lock on `scheduled_task` is released when the task body finishes. If two backends' cron timers fire for the same scheduled instant with a gap larger than the task body's runtime, both execute it. Penpot's cron entries are idempotent (`session-gc`, `objects-gc`, `storage-gc-*`, `tasks-gc`, `upload-session-gc`, `file-gc-scheduler`); the exceptions are `:telemetry` (would double-report) and `:audit-log-archive` (depends on archive target idempotency). - **`wrk/submit! ::dedupe true`** does a non-atomic `DELETE` then `INSERT`. Concurrent cross-backend submits can both bypass the `DELETE` (each sees the other's uncommitted insert as absent) and end up with duplicate `'new'` rows. Each row claims and runs once independently, so the underlying work is fine; the "at most one pending" guarantee weakens. Penpot in production lives with both: horizontal-scale deployments accept "exactly-once" as "essentially-once for idempotent operations." Devenv parallel instances handle it by running workers only on ws0 (see `mem:devenv/core`). diff --git a/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql b/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql index 68cf4af52a..3383b616d6 100644 --- a/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql +++ b/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql @@ -10,6 +10,13 @@ --- (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(), @@ -17,7 +24,7 @@ CREATE TABLE upload_session_chunk ( created_at timestamptz NOT NULL DEFAULT now(), session_id uuid NOT NULL REFERENCES upload_session(id) ON DELETE NO ACTION DEFERRABLE, - object_id uuid NOT NULL REFERENCES storage_object(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) diff --git a/backend/src/app/rpc/commands/media.clj b/backend/src/app/rpc/commands/media.clj index 3a0d920c4a..f476a48ad4 100644 --- a/backend/src/app/rpc/commands/media.clj +++ b/backend/src/app/rpc/commands/media.clj @@ -339,6 +339,8 @@ ;; --- Chunked Upload: Upload a single chunk +(declare ^:private check-upload-chunk-slot) + (def ^:private schema:upload-chunk [:map {:title "upload-chunk"} [:session-id ::sm/uuid] @@ -350,13 +352,69 @@ [: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} [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id session-id index content] :as _params}] - (let [session (db/get pool :upload-session {:id session-id :profile-id profile-id})] + (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 + :chunk (str index "/" (:total-chunks session)) + :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)) + 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}) + +(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 @@ -380,49 +438,22 @@ :size (:size content) :max-size (cf/get :upload-max-chunk-size))) - (when (db/get* pool :upload-session-chunk {:session-id session-id :chunk-index 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 :chunk-already-exists :hint "chunk already uploaded for this session and index" :session-id session-id :index index)) - (l/trc :hint "upload-chunk" - :session-id session-id - :chunk (str index "/" (:total-chunks session)) - :size (:size content) - :path (:path content)) + (db/insert! conn :upload-session-chunk + {:session-id session-id + :object-id nil + :chunk-index index}) - (let [storage (sto/resolve cfg) - data (sto/content (:path content)) - object (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})] - ;; NOTE: the pre-check above covers the common path, but two - ;; concurrent uploads of the same index can still race past it; the - ;; UNIQUE (session_id, chunk_index) constraint is the backstop. In - ;; that case the just-created storage object is left orphaned but - ;; touched, so touched-gc reclaims it. - (try - (db/insert! pool :upload-session-chunk - {:session-id session-id - :object-id (:id object) - :chunk-index index}) - (catch java.sql.SQLException cause - (if (db/duplicate-key-error? cause) - (ex/raise :type :validation - :code :chunk-already-exists - :hint "chunk already uploaded for this session and index" - :session-id session-id - :index index - :cause cause) - (throw cause)))))) - - {:session-id session-id - :index index}) + session)) ;; --- Chunked Upload: shared helpers diff --git a/backend/test/backend_tests/rpc_media_test.clj b/backend/test/backend_tests/rpc_media_test.clj index d8634dc252..b0b649ef9e 100644 --- a/backend/test/backend_tests/rpc_media_test.clj +++ b/backend/test/backend_tests/rpc_media_test.clj @@ -1105,47 +1105,6 @@ (db/exec! conn ["delete from profile where id = ?" (:id prof)]))))))))) -(t/deftest chunked-upload-duplicate-index-race-backstop - ;; Forces the UNIQUE backstop past the pre-check (simulates two concurrent - ;; uploads of the same index): the insert collides and the client still - ;; gets :validation/:chunk-already-exists. The just-created orphaned object - ;; stays touched so touched-gc reclaims it, and there is still exactly one - ;; mapping row. - (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} - upload {::th/type :upload-chunk - ::rpc/profile-id (:id prof) - :session-id session-id - :index 0 - :content mfile} - out1 (th/command! upload) - orig-get* @#'db/get*] - (t/is (nil? (:error out1))) - - (with-mocks [_mock {:target 'app.db/get* - ;; blind the duplicate pre-check, delegate the rest - :return (fn [ds table params & opts] - (if (= table :upload-session-chunk) - nil - (apply orig-get* ds table params opts)))}] - (let [before (:count (th/db-exec-one! ["select count(*) from storage_object"])) - out2 (th/command! upload)] - (t/is (some? (:error out2))) - (t/is (= :validation (-> out2 :error ex-data :type))) - (t/is (= :chunk-already-exists (-> out2 :error ex-data :code))) - ;; one orphaned object was created... - (t/is (= (inc before) (:count (th/db-exec-one! ["select count(*) from storage_object"])))) - ;; ...but still a single mapping row... - (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?" - session-id])))) - ;; ...and the orphan stays touched for touched-gc. - (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null and id not in (select object_id from upload_session_chunk)"])))))))) - ;; --- Clone File Media Object BOLA tests --- (defn- create-storage-object!