fix(webdav): make PATCH's concurrency guard a real compare-and-swap

The app-level ETag re-check before the write still left a gap between the check and the actual UPDATE for a concurrent writer to land in.
Push the check into the write path itself: swap_blob_hash now takes an expected_hash and only applies the SET under the same FOR UPDATE row lock it already held, closing the race instead of just narrowing it. Adds ErrorKind::PreconditionFailed (412) for the CAS-miss path; PUT/WOPI/chunked-upload keep blind-overwrite semantics by passing None

src/application/ports/file_ports.rs

@@ -99,6 +99,16 @@ pub trait FileUploadUseCase: Send + Sync + 'static {
     /// on the overwrite branch and `authz.require(caller, Create,
     /// Folder|Drive(id))` on the new-file branch. Handlers just plumb
     /// `caller_id` through — no protocol-layer authz.
+    ///
+    /// `expected_hash`: forwarded to
+    /// `FileWritePort::update_file_content_with_blob` on the overwrite
+    /// branch for compare-and-swap; ignored on the new-file branch
+    /// (nothing to compare against). Pass `None` for plain PUT/WOPI/
+    /// chunked-upload last-write-wins semantics; pass the pre-write
+    /// snapshot's content hash for PATCH, where a concurrent write
+    /// during the (potentially slow) splice must be rejected rather
+    /// than silently clobbered.
+    #[allow(clippy::too_many_arguments)]
     async fn update_file_streaming_with_perms(
         &self,
         path: &str,
@@ -107,6 +117,7 @@ pub trait FileUploadUseCase: Send + Sync + 'static {
         content_type: &str,
         modified_at: Option<i64>,
         caller_id: Uuid,
+        expected_hash: Option<&str>,
     ) -> Result<FileDto, DomainError>;
 }
 

src/application/ports/storage_ports.rs

@@ -299,6 +299,14 @@ pub trait FileWritePort: Send + Sync + 'static {
     ///
     /// `caller_id` is stamped into `updated_by` alongside the
     /// `updated_at` bump (§14 provenance).
+    ///
+    /// `expected_hash`: when `Some`, makes this a true compare-and-swap —
+    /// the write only takes effect if the row's current `blob_hash`
+    /// still equals it, checked and applied atomically under the same
+    /// row lock (no gap between check and write for a concurrent writer
+    /// to land in). A mismatch returns `ErrorKind::PreconditionFailed`
+    /// and leaves the row untouched. `None` keeps the previous
+    /// blind-overwrite behaviour (PUT/WOPI/chunked-upload finalize).
     async fn update_file_content_with_blob(
         &self,
         file_id: &str,
@@ -306,6 +314,7 @@ pub trait FileWritePort: Send + Sync + 'static {
         size: u64,
         modified_at: Option<i64>,
         caller_id: Uuid,
+        expected_hash: Option<&str>,
     ) -> Result<(String, i64), DomainError>;
 
     /// Registers file metadata WITHOUT writing content to disk (write-behind).

src/application/services/file_upload_service.rs

@@ -305,7 +305,7 @@ impl FileUploadService {
         let file = file_read.get_file(file_id).await?;
         let (new_hash, updated_at) = self
             .file_write
-            .update_file_content_with_blob(file_id, &blob.hash, blob.size, None, caller_id)
+            .update_file_content_with_blob(file_id, &blob.hash, blob.size, None, caller_id, None)
             .await?;
         // The file maps to a different blob now — stale cached content must
         // never be served for the rest of its TTI window.
@@ -506,6 +506,7 @@ impl FileUploadUseCase for FileUploadService {
     /// member and cross-tenant PUT. See
     /// `docs/plan/authz_audit/nextcloud.md` and the sibling native
     /// `/webdav/*` handler.
+    #[allow(clippy::too_many_arguments)]
     async fn update_file_streaming_with_perms(
         &self,
         path: &str,
@@ -514,6 +515,7 @@ impl FileUploadUseCase for FileUploadService {
         content_type: &str,
         modified_at: Option<i64>,
         caller_id: Uuid,
+        expected_hash: Option<&str>,
     ) -> Result<FileDto, DomainError> {
         let Some(authz) = &self.authorization else {
             return Err(DomainError::internal_error(
@@ -552,6 +554,7 @@ impl FileUploadUseCase for FileUploadService {
                     blob.size,
                     modified_at,
                     caller_id,
+                    expected_hash,
                 )
                 .await?;
             // Invalidate content cache — file content has changed.

src/application/services/trash_service_test.rs

@@ -589,6 +589,7 @@ impl FileWritePort for MockFileRepository {
         _size: u64,
         _modified_at: Option<i64>,
         _caller_id: Uuid,
+        _expected_hash: Option<&str>,
     ) -> std::result::Result<(String, i64), DomainError> {
         Ok((String::new(), 0))
     }

src/common/stubs.rs

@@ -196,6 +196,7 @@ impl FileWritePort for StubFileWritePort {
         _size: u64,
         _modified_at: Option<i64>,
         _caller_id: Uuid,
+        _expected_hash: Option<&str>,
     ) -> Result<(String, i64), DomainError> {
         Ok((String::new(), 0))
     }
@@ -491,6 +492,7 @@ impl FileUploadUseCase for StubFileUploadUseCase {
         Ok(FileDto::default())
     }
 
+    #[allow(clippy::too_many_arguments)]
     async fn update_file_streaming_with_perms(
         &self,
         _path: &str,
@@ -499,6 +501,7 @@ impl FileUploadUseCase for StubFileUploadUseCase {
         _content_type: &str,
         _modified_at: Option<i64>,
         _caller_id: Uuid,
+        _expected_hash: Option<&str>,
     ) -> Result<FileDto, DomainError> {
         Ok(FileDto::default())
     }

src/domain/errors.rs

@@ -39,6 +39,12 @@ pub enum ErrorKind {
     /// `AlreadyExists` (which is a uniqueness violation) so audit
     /// readers can tell them apart.
     Conflict,
+    /// RFC 7232 precondition failure — a caller-supplied conditional
+    /// (If-Match, or an internal compare-and-swap standing in for one)
+    /// did not hold against the resource's current state. Maps to
+    /// HTTP 412. Distinct from `Conflict` (409): this is specifically
+    /// "the state you thought you were writing against has moved."
+    PreconditionFailed,
 }
 
 impl ErrorKind {
@@ -58,6 +64,7 @@ impl ErrorKind {
             ErrorKind::DatabaseError => "Database Error",
             ErrorKind::QuotaExceeded => "Quota Exceeded",
             ErrorKind::Conflict => "Conflict",
+            ErrorKind::PreconditionFailed => "Precondition Failed",
         }
     }
 }
@@ -196,6 +203,17 @@ impl DomainError {
         }
     }
 
+    /// Creates a precondition-failed error (RFC 7232 / CAS mismatch)
+    pub fn precondition_failed<S: Into<String>>(entity_type: &'static str, message: S) -> Self {
+        Self {
+            kind: ErrorKind::PreconditionFailed,
+            entity_type,
+            entity_id: None,
+            message: message.into(),
+            source: None,
+        }
+    }
+
     /// Creates a validation error
     pub fn validation_error<S: Into<String>>(message: S) -> Self {
         Self {

src/infrastructure/repositories/pg/file_blob_write_repository.rs

@@ -153,6 +153,15 @@ impl FileBlobWriteRepository {
     /// row — not the row's owner. D2 shared drives let non-owners
     /// overwrite content; the previous `updated_by = f.user_id` would
     /// have silently recorded the wrong principal.
+    /// `expected_hash`, when `Some`, turns this into a real
+    /// compare-and-swap: the SET clause only takes effect if the row's
+    /// `blob_hash` still matches at the moment the `FOR UPDATE` lock is
+    /// held (same statement, same transaction — no gap a concurrent
+    /// writer can land in). A mismatch leaves the row untouched and is
+    /// reported back via the `matched` flag rather than silently
+    /// overwriting a sibling PATCH's content. `None` preserves the old
+    /// blind-overwrite behaviour for PUT/WOPI/chunked-upload finalize,
+    /// where last-write-wins is the intended HTTP semantics.
     async fn swap_blob_hash(
         &self,
         file_id: &str,
@@ -160,57 +169,83 @@ impl FileBlobWriteRepository {
         new_size: i64,
         modified_at: Option<i64>,
         caller_id: Uuid,
+        expected_hash: Option<&str>,
     ) -> Result<(String, i64), DomainError> {
-        // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
+        // Atomic CTE: capture old hash then conditionally update in one
+        // round-trip, no TOCTOU. The CASE arms make the SET a no-op when
+        // `expected_hash` is given and doesn't match `old.blob_hash` —
+        // the row is still returned (with its unchanged values) so the
+        // caller can tell "mismatch" apart from "file not found".
         // Deadlock victims (40P01) retry before the compensation below runs —
         // a successful retry must keep the new blob reference alive.
-        let (old_hash, updated_at) = match retry_on_deadlock("files.swap_blob_hash", || {
-            sqlx::query_as::<_, (String, i64)>(
-                r#"
+        let (old_hash, updated_at, matched) =
+            match retry_on_deadlock("files.swap_blob_hash", || {
+                sqlx::query_as::<_, (String, i64, bool)>(
+                    r#"
                 WITH old AS (
                     SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
                 )
                 UPDATE storage.files f
-                   SET blob_hash = $1, size = $2,
-                       updated_at = COALESCE(to_timestamp($4), NOW()),
-                       updated_by = $5
+                   SET blob_hash = CASE WHEN $6::text IS NULL OR old.blob_hash = $6
+                                        THEN $1 ELSE f.blob_hash END,
+                       size = CASE WHEN $6::text IS NULL OR old.blob_hash = $6
+                                   THEN $2 ELSE f.size END,
+                       updated_at = CASE WHEN $6::text IS NULL OR old.blob_hash = $6
+                                         THEN COALESCE(to_timestamp($4), NOW()) ELSE f.updated_at END,
+                       updated_by = CASE WHEN $6::text IS NULL OR old.blob_hash = $6
+                                         THEN $5 ELSE f.updated_by END
                   FROM old
                  WHERE f.id = old.id
-                RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint
+                RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint,
+                          ($6::text IS NULL OR old.blob_hash = $6)
                 "#,
-            )
-            .bind(new_hash)
-            .bind(new_size)
-            .bind(file_id)
-            .bind(modified_at.map(|t| t as f64))
-            .bind(caller_id)
-            .fetch_optional(self.pool.as_ref())
-        })
-        .await
-        {
-            Ok(Some(row)) => row,
-            Ok(None) => {
-                // File not found — compensate: remove the new blob ref
-                if let Err(e) = self.dedup.remove_reference(new_hash).await {
-                    tracing::error!("Blob orphaned after missing file: {}", e);
+                )
+                .bind(new_hash)
+                .bind(new_size)
+                .bind(file_id)
+                .bind(modified_at.map(|t| t as f64))
+                .bind(caller_id)
+                .bind(expected_hash)
+                .fetch_optional(self.pool.as_ref())
+            })
+            .await
+            {
+                Ok(Some(row)) => row,
+                Ok(None) => {
+                    // File not found — compensate: remove the new blob ref
+                    if let Err(e) = self.dedup.remove_reference(new_hash).await {
+                        tracing::error!("Blob orphaned after missing file: {}", e);
+                    }
+                    return Err(DomainError::not_found("File", file_id));
                 }
-                return Err(DomainError::not_found("File", file_id));
-            }
-            Err(e) => {
-                // UPDATE failed — compensate: remove the new blob ref
-                if let Err(rollback_err) = self.dedup.remove_reference(new_hash).await {
-                    tracing::error!(
-                        "Blob orphaned after failed UPDATE — hash: {}, err: {}",
-                        &new_hash[..12],
-                        rollback_err
-                    );
+                Err(e) => {
+                    // UPDATE failed — compensate: remove the new blob ref
+                    if let Err(rollback_err) = self.dedup.remove_reference(new_hash).await {
+                        tracing::error!(
+                            "Blob orphaned after failed UPDATE — hash: {}, err: {}",
+                            &new_hash[..12],
+                            rollback_err
+                        );
+                    }
+                    return Err(DomainError::internal_error(
+                        "FileBlobWrite",
+                        format!("update: {e}"),
+                    ));
                 }
-                return Err(DomainError::internal_error(
-                    "FileBlobWrite",
-                    format!("update: {e}"),
-                ));
+            };
+
+        if !matched {
+            // CAS lost the race — some other writer's content is now the
+            // row's truth. Release the blob we ingested for nothing;
+            // nothing was written.
+            if let Err(e) = self.dedup.remove_reference(new_hash).await {
+                tracing::error!("Blob orphaned after CAS mismatch: {}", e);
             }
-        };
+            return Err(DomainError::precondition_failed(
+                "File",
+                "content was modified concurrently",
+            ));
+        }
 
         // Decrement old blob ref (only if hash changed, best-effort)
         if old_hash != new_hash
@@ -790,12 +825,20 @@ impl FileWritePort for FileBlobWriteRepository {
         size: u64,
         modified_at: Option<i64>,
         caller_id: Uuid,
+        expected_hash: Option<&str>,
     ) -> Result<(String, i64), DomainError> {
         // The content was already ingested into the chunk store by the
         // upload-ingest layer; swap_blob_hash consumes its reference and
         // releases it on failure.
         let swapped = self
-            .swap_blob_hash(file_id, blob_hash, size as i64, modified_at, caller_id)
+            .swap_blob_hash(
+                file_id,
+                blob_hash,
+                size as i64,
+                modified_at,
+                caller_id,
+                expected_hash,
+            )
             .await?;
         // The file now maps to a different blob — drop the read-side cache
         // entry so streaming downloads cannot serve the previous content

src/interfaces/api/handlers/webdav_handler.rs

@@ -1972,6 +1972,7 @@ async fn handle_put(
             &content_type,
             None,
             user.id,
+            None,
         )
         .await;
 
@@ -2303,25 +2304,15 @@ async fn handle_patch(
         ));
     }
 
-    // ── Optimistic-concurrency re-check ───────────────────────────────
-    // `file.etag` was snapshotted before the (potentially slow) splice +
-    // CAS-ingest above. Re-verify nothing else wrote to this file in the
-    // meantime, narrowing the window in which two concurrent PATCHes to
-    // disjoint ranges — each individually passing its own If-Match check
-    // against the same stale snapshot — could otherwise silently clobber
-    // each other on the blind-overwrite write path below.
-    if let Ok(current) = file_retrieval_service
-        .get_file_by_path(&path, drive_id)
-        .await
-        && current.etag != file.etag
-    {
-        upload_ingest::discard_ingested(&state.core.dedup_service, &ingested).await;
-        return Err(AppError::precondition_failed(
-            "File was modified concurrently — retry the PATCH",
-        ));
-    }
-
-    // ── Atomic store ──────────────────────────────────────────────────
+    // ── Atomic store, compare-and-swap on the pre-splice content hash ──
+    // `file.content_hash` was snapshotted before the (potentially slow)
+    // splice + CAS-ingest above. Passing it as `expected_hash` makes the
+    // write itself a compare-and-swap: the repository checks and applies
+    // under the same row lock, so nothing else can write to this file
+    // between the check and the write. This is what actually closes the
+    // race two concurrent PATCHes to disjoint ranges could otherwise hit
+    // — each individually passing its own If-Match check against the
+    // same stale snapshot, then blindly overwriting each other.
     let new_size = ingested.size;
     let content_type = ingested.content_type.clone();
     let result = file_upload_service
@@ -2332,6 +2323,7 @@ async fn handle_patch(
             &content_type,
             None,
             user.id,
+            Some(&file.content_hash),
         )
         .await;
 

src/interfaces/api/handlers/wopi_handler.rs

@@ -413,6 +413,7 @@ async fn put_file(
             &content_type,
             None,
             claims_sub_uuid,
+            None,
         )
         .await;
 

src/interfaces/errors.rs

@@ -131,6 +131,7 @@ impl From<DomainError> for AppError {
             ErrorKind::DatabaseError => StatusCode::INTERNAL_SERVER_ERROR,
             ErrorKind::QuotaExceeded => StatusCode::INSUFFICIENT_STORAGE,
             ErrorKind::Conflict => StatusCode::CONFLICT,
+            ErrorKind::PreconditionFailed => StatusCode::PRECONDITION_FAILED,
         };
 
         Self {

src/interfaces/nextcloud/uploads_handler.rs

@@ -466,6 +466,10 @@ async fn handle_assemble(
     // AuthZ audit #2 (2026-07-12): route DomainError through
     // `AppError::from` so authz denials keep the graduated 403/404
     // shape instead of collapsing into 500.
+    //
+    // No client-supplied ETag to enforce here (NC chunked MOVE has no
+    // If-Match semantics) — `expected_hash: None`, same as every other
+    // plain-write callsite; only PATCH's CAS passes `Some(&hash)`.
     let dto = match upload_service
         .update_file_streaming_with_perms(
             &internal_path,
@@ -474,6 +478,7 @@ async fn handle_assemble(
             &content_type,
             oc_mtime,
             user.id,
+            None,
         )
         .await
     {

src/interfaces/nextcloud/webdav_handler.rs

@@ -912,6 +912,7 @@ async fn handle_put(
             &content_type,
             oc_mtime,
             session.user.id,
+            None,
         )
         .await
         .map_err(AppError::from)?;
@@ -1145,25 +1146,15 @@ async fn handle_patch(
         ));
     }
 
-    // ── Optimistic-concurrency re-check ───────────────────────────────
-    // `file.etag` was snapshotted before the (potentially slow) splice +
-    // CAS-ingest above. Re-verify nothing else wrote to this file in the
-    // meantime, narrowing the window in which two concurrent PATCHes to
-    // disjoint ranges — each individually passing its own If-Match check
-    // against the same stale snapshot — could otherwise silently clobber
-    // each other on the blind-overwrite write path below.
-    if let Ok(current) = file_service
-        .get_file_by_path(&internal_path, chroot.drive_id)
-        .await
-        && current.etag != file.etag
-    {
-        discard_ingested(&state.core.dedup_service, &ingested).await;
-        return Err(AppError::precondition_failed(
-            "File was modified concurrently — retry the PATCH",
-        ));
-    }
-
-    // ── Atomic store ──────────────────────────────────────────────────
+    // ── Atomic store, compare-and-swap on the pre-splice content hash ──
+    // `file.content_hash` was snapshotted before the (potentially slow)
+    // splice + CAS-ingest above. Passing it as `expected_hash` makes the
+    // write itself a compare-and-swap: the repository checks and applies
+    // under the same row lock, so nothing else can write to this file
+    // between the check and the write. This is what actually closes the
+    // race two concurrent PATCHes to disjoint ranges could otherwise hit
+    // — each individually passing its own If-Match check against the
+    // same stale snapshot, then blindly overwriting each other.
     let new_size = ingested.size;
     let content_type = ingested.content_type.clone();
     let stored = upload_service
@@ -1174,6 +1165,7 @@ async fn handle_patch(
             &content_type,
             None,
             session.user.id,
+            Some(&file.content_hash),
         )
         .await
         .map_err(AppError::from)?;