♻️ 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

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

.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`).

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)

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
 

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!