fix(backend): enhance storage/resource concurrency, idempotency, and deletion safety

Summary of changes:

- Resources Bind Idempotency (High 1+2):
  - Check existing active storage object bindings before duplicate check.
  - Return existing binding (`reused: true`) on retry with same upload_id.
  - Filter by `status == "active"` to bypass dead/deleted rows during reuse check.

- Storage Concurrent Overwrite (High 3):
  - Add `acquire_named_lock` and `release_named_lock` helpers using MySQL `GET_LOCK`/`RELEASE_LOCK` hashed to <= 64 chars.
  - Wrap `upload_bytes_to_session` PUT+INSERT critical section with named lock on `object_key`.
  - Re-check key collision inside lock; append ULID suffix on collision.

- Shared Reference Deletion Protection (Medium 4):
  - Check active references before deleting storage objects in `delete_resource`.
  - Delete only `DataResources` record if storage object is still referenced elsewhere.

- Robust Usage Type Fallback (Medium 5):
  - Replace direct dict lookup for `USAGE_TYPE_TO_PURPOSE[item.usage_type]` with `.get(..., "workspace")` default.

- Idempotency Key Path Matching (Medium 6):
  - Move `file_name`/`target_path` validation forward and include path dimension in comparison.
  - Strip uniqueness suffix via `_strip_uniqueness_suffix` before key comparison to avoid false 409s on valid retries.

- Usage Type & Bind Concurrency Control (Low 7 & 8):
  - Reject bind requests with 409 if upload session purpose is not `data_resource`.
  - Wrap resource duplicate check and creation in named lock using `(owner, directory, name)`.

- Trash Key Uniqueness & Restore Compatibility (Low 9):
  - Update `trash_key` format to `{purpose}/{object_key}-{storage_object_id}` to prevent collisions.
  - Update `object_key_hash` on trash move.
  - Update restore logic in `storage_api.py` to strip suffix while maintaining backward compatibility with legacy keys.

- Dead Code Removal (Low 10):
  - Remove unreachable `upload_status = "failed"` and redundant `session.rollback()` in `IntegrityError` block.

- Tests & Mocks:
  - Add/update 5 test cases covering non-data_resource bind rejection, suffix stripping, and path recovery.
  - Add named lock statement mocks for DB testing.

backend/src/backend/resources.py

@@ -13,7 +13,7 @@ from common.storage.schemas import (
     DownloadUrlRequest,
 )
 from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.dependencies import (
@@ -28,8 +28,10 @@ from backend.schemas import (
     ResourceRelativePathRequest,
 )
 from backend.services.storage import (
+    acquire_named_lock,
     create_download_url_payload,
     create_upload_record,
+    release_named_lock,
     soft_delete_object,
     upload_bytes_to_session,
 )
@@ -250,12 +252,34 @@ async def bind_resource(
             status.HTTP_403_FORBIDDEN,
             "upload belongs to another user",
         )
+    if upload.usage_type != "data_resource":
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            "upload was not created for a data resource",
+        )
     item = await session.get(StorageObjects, upload.storage_object_id)
     if item is None:
         raise HTTPException(
             status.HTTP_409_CONFLICT,
             "storage object metadata is missing",
         )
+
+    # 幂等重试:同一 storage object 已有 active 的资源行,直接返回,
+    # 不重复创建、不触发同名查重。已删除的旧绑定行(status != active)
+    # 不会命中,走下方新建流程。
+    existing = await session.scalar(
+        select(DataResources).where(
+            DataResources.storage_object_id == item.storage_object_id,
+            DataResources.status == "active",
+        )
+    )
+    if existing is not None:
+        return {
+            "request_id": context.request_id,
+            "data": resource_payload(existing, item),
+            "meta": {"reused": True},
+        }
+
     # Persist resource_name / description / visibility override.
     item.visibility = payload.visibility
     # (description lives on DataResources, not on StorageObjects.)
@@ -267,6 +291,14 @@ async def bind_resource(
         context.workspace.workspace_id,
         context.user.user_id,
     )
+    # 查重 + 建行没有唯一索引兜底,按 (owner, 目录, 名称) 加命名锁,
+    # 避免两个并发 bind 同时通过检查产生重复行。
+    lock_name = await acquire_named_lock(
+        session,
+        f"bind-resource:{context.workspace.workspace_id}:"
+        f"{context.user.user_id}:{new_directory}:{payload.resource_name}",
+    )
+    try:
         same_name_rows = (
             await session.execute(
                 select(DataResources, StorageObjects)
@@ -280,7 +312,6 @@ async def bind_resource(
                     DataResources.owner_user_id == context.user.user_id,
                     DataResources.resource_name == payload.resource_name,
                     DataResources.status == "active",
-                DataResources.storage_object_id != item.storage_object_id,
                 )
             )
         ).all()
@@ -296,13 +327,6 @@ async def bind_resource(
                     "a data resource with this name already exists in this directory",
                 )
 
-    existing = await session.scalar(
-        select(DataResources).where(
-            DataResources.storage_object_id == item.storage_object_id
-        )
-    )
-    reused = existing is not None
-    if existing is None:
         existing = DataResources(
             resource_id=new_ulid(),
             workspace_id=context.workspace.workspace_id,
@@ -316,10 +340,12 @@ async def bind_resource(
         session.add(existing)
         await session.flush()
         await session.refresh(existing)
+    finally:
+        await release_named_lock(session, lock_name)
     return {
         "request_id": context.request_id,
         "data": resource_payload(existing, item),
-        "meta": {"reused": reused},
+        "meta": {"reused": False},
     }
 
 
@@ -461,6 +487,16 @@ async def delete_resource(
             status.HTTP_403_FORBIDDEN,
             "resource can only be deleted by its owner or an administrator",
         )
+    # 底层 storage object 可能被其他 active 资源行共享;仍有引用时
+    # 只删 DataResources 行,不动底层对象。
+    shared_count = await session.scalar(
+        select(func.count(DataResources.resource_id)).where(
+            DataResources.storage_object_id == resource.storage_object_id,
+            DataResources.resource_id != resource.resource_id,
+            DataResources.status == "active",
+        )
+    )
+    if not shared_count:
         await soft_delete_object(resource.storage_object_id, request, session)
     resource.status = "deleted"
     resource.deleted_at = datetime.now(UTC).replace(tzinfo=None)

backend/src/backend/services/storage.py

@@ -45,7 +45,7 @@ from common.storage.schemas import (
     ServerObjectRequest,
 )
 from fastapi import HTTPException, Request, status
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -98,6 +98,59 @@ def _hash_bytes(value: str) -> bytes:
     return _h.sha256(value.encode("utf-8")).digest()
 
 
+def _object_key_tail(object_key: str, workspace_id: str, user_id: str) -> str:
+    """object_key 去掉 ``{ws_id}/{user_id}/`` 前缀后的相对路径部分。"""
+    prefix = f"{workspace_id}/{user_id}/"
+    if object_key.startswith(prefix):
+        return object_key[len(prefix):]
+    return object_key
+
+
+def _strip_uniqueness_suffix(tail: str) -> str:
+    """去掉 _resolve_unique_object_key 追加的 ``-<ulid>`` 后缀,恢复请求时的原始路径。"""
+    dir_part, sep, name = tail.rpartition("/")
+    stem, dot, ext = name.rpartition(".")
+    if not dot:
+        stem, ext = name, ""
+    base, dash, suffix = stem.rpartition("-")
+    if dash and len(suffix) == 26 and suffix.isalnum():
+        stem = base
+    name = f"{stem}.{ext}" if ext else stem
+    return f"{dir_part}{sep}{name}"
+
+
+async def acquire_named_lock(
+    session: AsyncSession,
+    name: str,
+    *,
+    timeout_seconds: int = 10,
+) -> str:
+    """获取 MySQL 命名锁(绑定当前会话连接),返回实际锁名。
+
+    MySQL 锁名上限 64 字符,统一哈希压缩。调用方必须在同一 session 上
+    用 release_named_lock 释放 —— 连接归还连接池时锁不会自动释放。
+    """
+    digest = hashlib.sha256(name.encode("utf-8")).hexdigest()
+    lock_name = f"mp:{digest[:61]}"
+    acquired = await session.scalar(
+        text("SELECT GET_LOCK(:name, :timeout)").bindparams(
+            name=lock_name, timeout=timeout_seconds
+        )
+    )
+    if acquired != 1:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "failed to acquire storage lock; retry later",
+        )
+    return lock_name
+
+
+async def release_named_lock(session: AsyncSession, lock_name: str) -> None:
+    await session.execute(
+        text("SELECT RELEASE_LOCK(:name)").bindparams(name=lock_name)
+    )
+
+
 async def _resolve_unique_object_key(
     session: AsyncSession,
     object_key: str,
@@ -215,16 +268,40 @@ async def create_upload_record(
     stored_key = normalized_idempotency_key(
         payload.workspace_id, payload.user_id, payload.idempotency_key
     )
+    safe_name = _safe_file_name(payload.file_name)
+    # Defense-in-depth: schemas already validate target_path, but the
+    # helper is called directly from some callers so re-validate here.
+    target_path = (payload.target_path or "").strip("/")
+    if target_path and (target_path.startswith("/") or "\\" in target_path
+                        or any(seg == ".." for seg in target_path.split("/"))
+                        or any(ord(c) < 0x20 or ord(c) == 0x7F for c in target_path)):
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            "target_path must be a clean relative POSIX path",
+        )
+    clean_target = "/".join(
+        _safe_path_segment(seg) for seg in target_path.split("/") if seg
+    )
+    requested_tail = f"{clean_target}/{safe_name}" if clean_target else safe_name
+
     existing = await session.scalar(
         select(UploadSessions).where(UploadSessions.idempotency_key == stored_key)
     )
     if existing is not None:
+        # 已有会话的 object_key 可能带唯一化后缀,比较前恢复原始路径,
+        # 保证同一文件的幂等重试通过、不同路径/文件名的复用报 409。
+        existing_tail = _strip_uniqueness_suffix(
+            _object_key_tail(
+                existing.object_key, payload.workspace_id, payload.user_id
+            )
+        )
         if (
             existing.workspace_id != payload.workspace_id
             or existing.user_id != payload.user_id
             or existing.expected_size_bytes != payload.expected_size_bytes
             or existing.expected_hash != payload.expected_hash
             or existing.content_type != payload.content_type
+            or existing_tail != requested_tail
         ):
             raise HTTPException(
                 status.HTTP_409_CONFLICT,
@@ -240,24 +317,9 @@ async def create_upload_record(
             payload.usage_type,
             workspace_artifact_bucket=workspace.artifact_bucket,
         )
-        safe_name = _safe_file_name(payload.file_name)
+        object_key = (
-        # Defense-in-depth: schemas already validate target_path, but the
+            f"{payload.workspace_id}/{payload.user_id}/{requested_tail}"
-        # helper is called directly from some callers so re-validate here.
-        target_path = (payload.target_path or "").strip("/")
-        if target_path and (target_path.startswith("/") or "\\" in target_path
-                            or any(seg == ".." for seg in target_path.split("/"))
-                            or any(ord(c) < 0x20 or ord(c) == 0x7F for c in target_path)):
-            raise HTTPException(
-                status.HTTP_422_UNPROCESSABLE_ENTITY,
-                "target_path must be a clean relative POSIX path",
         )
-        clean_target = "/".join(
-            _safe_path_segment(seg) for seg in target_path.split("/") if seg
-        )
-        if clean_target:
-            object_key = f"{payload.workspace_id}/{payload.user_id}/{clean_target}/{safe_name}"
-        else:
-            object_key = f"{payload.workspace_id}/{payload.user_id}/{safe_name}"
         object_key = await _resolve_unique_object_key(session, object_key)
         upload = UploadSessions(
             upload_id=new_ulid(),
@@ -342,6 +404,29 @@ async def upload_bytes_to_session(
         upload.upload_status = "expired"
         raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
 
+    # 按 object_key 串行化「PUT + INSERT」临界区:否则两个同 key 并发上传
+    # 会互相覆盖字节,后 INSERT 的一方 409 时字节已被污染。
+    lock_name = await acquire_named_lock(
+        session, f"upload:{upload.bucket_name}:{upload.object_key}"
+    )
+    try:
+        # 锁内复查:创建会话后该 key 可能已被其他对象占用,必要时换 key。
+        collision = await session.scalar(
+            select(StorageObjects.storage_object_id).where(
+                StorageObjects.bucket_name == upload.bucket_name,
+                StorageObjects.object_key_hash == upload.object_key_hash,
+                StorageObjects.object_status == "available",
+                StorageObjects.is_deleted == 0,
+            )
+        )
+        if collision is not None:
+            # 新 key 带随机 ULID 后缀,不会与其他并发会话冲突,无需重新加锁。
+            new_key = await _resolve_unique_object_key(
+                session, upload.object_key
+            )
+            upload.object_key = new_key
+            upload.object_key_hash = _hash_bytes(new_key)
+
         content = await request.body()
         actual_size = len(content)
 
@@ -395,8 +480,8 @@ async def upload_bytes_to_session(
         try:
             await session.flush()
         except IntegrityError as exc:
-        upload.upload_status = "failed"
+            # 锁内已复查,走到这里说明发生了极罕见的跨锁竞争;
-        await session.rollback()
+            # 事务会由上层回滚,这里只需报错。
             raise HTTPException(
                 status.HTTP_409_CONFLICT,
                 "a file with this name already exists at this path; rename and retry",
@@ -406,6 +491,8 @@ async def upload_bytes_to_session(
         upload.upload_status = "completed"
         upload.completed_at = _utcnow_naive()
         return item
+    finally:
+        await release_named_lock(session, lock_name)
 
 
 # ── create_server_object_payload ────────────────────────────────────────
@@ -594,8 +681,9 @@ async def soft_delete_object(
             }
         }
     if item.storage_backend == settings.storage_backend and item.bucket_name and item.object_key:
-        source_purpose = USAGE_TYPE_TO_PURPOSE[item.usage_type]
+        source_purpose = USAGE_TYPE_TO_PURPOSE.get(item.usage_type, "workspace")
-        trash_key = f"{source_purpose}/{item.object_key}"
+        # 尾部拼 storage_object_id,防止同名文件多次删除在回收站互相覆盖。
+        trash_key = f"{source_purpose}/{item.object_key}-{item.storage_object_id}"
         trash_bucket = actual_bucket_name("trash")
         try:
             object_stores = request.app.state.object_stores
@@ -610,6 +698,7 @@ async def soft_delete_object(
         item.trash_key = trash_key
         item.bucket_name = trash_bucket
         item.object_key = trash_key
+        item.object_key_hash = _hash_bytes(trash_key)
         item.storage_uri = build_storage_uri(trash_bucket, trash_key)
     item.object_status = "deleted"
     item.deleted_at = _utcnow_naive()

backend/src/backend/storage_api.py

@@ -467,9 +467,17 @@ async def restore_object(
         # Cross-backend copy: get from trash, put back to source bucket.
         object_stores = request.app.state.object_stores
         data = await object_stores[actual_bucket_name("trash")].get(item.object_key)
-        source_purpose, _, source_key = item.object_key.partition("/")
+        source_purpose, _, trash_tail = item.object_key.partition("/")
         if not source_purpose:
             source_purpose = "workspace"
+        # trash key 尾部带 ``-{storage_object_id}`` 后缀(防同名覆盖),
+        # 恢复时剥掉;旧数据没有该后缀,endswith 判断天然兼容。
+        id_suffix = f"-{item.storage_object_id}"
+        source_key = (
+            trash_tail[: -len(id_suffix)]
+            if trash_tail.endswith(id_suffix)
+            else trash_tail
+        )
         target_bucket = actual_bucket_name(source_purpose)
         await object_stores[target_bucket].put(source_key, data)
     except Exception as exc:
@@ -481,6 +489,7 @@ async def restore_object(
     item.deleted_at = None
     item.bucket_name = target_bucket
     item.object_key = source_key
+    item.object_key_hash = hash_bytes(source_key)
     item.storage_uri = build_storage_uri(target_bucket, source_key)
     # Keep trash_key so the reaper can clean up the duplicate on its
     # next pass; we don't try to delete it here because a partial

backend/tests/test_resources.py

@@ -15,7 +15,12 @@ from backend.resources import (
     resource_directory,
     resource_payload,
 )
-from backend.services.storage import _safe_file_name, _safe_path_segment
+from backend.services.storage import (
+    _object_key_tail,
+    _safe_file_name,
+    _safe_path_segment,
+    _strip_uniqueness_suffix,
+)
 
 
 _BIND_WS = "01WS0000000000000000000A"
@@ -38,21 +43,33 @@ class _BindSessionMock:
     in the workspace with the same resource_name.
     """
 
-    def __init__(self, new_object_key: str, same_name_rows=(), reused_resource=None):
+    def __init__(
+        self,
+        new_object_key: str,
+        same_name_rows=(),
+        reused_resource=None,
+        usage_type: str = "data_resource",
+    ):
         self._new_object_key = new_object_key
         self._same_name_rows = list(same_name_rows)
         self._reused_resource = reused_resource
+        self._usage_type = usage_type
         self._scalar_calls = 0
 
     async def scalar(self, _stmt):
+        from sqlalchemy import TextClause
+
+        if isinstance(_stmt, TextClause):
+            return 1  # GET_LOCK 成功
         self._scalar_calls += 1
-        # 1st scalar: UploadSessions lookup; 2nd: storage_object_id reuse check.
+        # 1st scalar: UploadSessions lookup; 2nd: active 绑定行复用检查。
         if self._scalar_calls == 1:
             return SimpleNamespace(
                 upload_id="01UPL0000000000000000000B",
                 storage_object_id="01OBJ0000000000000000000B",
                 workspace_id=_BIND_WS,
                 user_id=_BIND_USER,
+                usage_type=self._usage_type,
             )
         if self._scalar_calls == 2 and self._reused_resource is not None:
             return self._reused_resource
@@ -62,6 +79,10 @@ class _BindSessionMock:
         return _make_storage_object(self._new_object_key)
 
     async def execute(self, _stmt):
+        from sqlalchemy import TextClause
+
+        if isinstance(_stmt, TextClause):
+            return _ExecuteResult([])  # RELEASE_LOCK
         # 模拟 SQL 的 owner 过滤与 self-exclusion:只返回属于当前用户且
         # 不是当前 upload 已绑定对象的同名候选。
         rows = [
@@ -222,6 +243,27 @@ async def test_bind_resource_allows_rebinding_same_storage_object() -> None:
     assert result["data"]["resource_name"] == "data.csv"
 
 
+@pytest.mark.asyncio
+async def test_bind_resource_rejects_non_data_resource_upload() -> None:
+    """其他用途(如 working_copy)的 upload session 不能 bind 成数据资源。"""
+    from backend.resources import bind_resource
+
+    session = _BindSessionMock(
+        new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
+        usage_type="working_copy",
+    )
+    with pytest.raises(Exception) as exc_info:
+        await bind_resource(
+            upload_id="01UPL0000000000000000000B",
+            payload=_bind_payload(),
+            request=MagicMock(),
+            context=_bind_context(),
+            session=session,
+        )
+    assert exc_info.value.status_code == 409
+    assert "not created for a data resource" in exc_info.value.detail
+
+
 def test_safe_path_segment_cleans_special_characters():
     assert _safe_path_segment("train") == "train"
     assert _safe_path_segment("train v1") == "train_v1"
@@ -328,6 +370,26 @@ def test_resource_directory_parses_object_key():
     assert resource_directory("other-bucket-key.csv", ws, user) == ""
 
 
+def test_object_key_tail_and_strip_uniqueness_suffix():
+    ws = "01WS00000000000000000000A"
+    user = "01USR000000000000000000A"
+    # 前缀剥离。
+    assert _object_key_tail(f"{ws}/{user}/sub/data.csv", ws, user) == "sub/data.csv"
+    assert _object_key_tail("no-prefix.csv", ws, user) == "no-prefix.csv"
+    # 带唯一化后缀的 key 恢复为原始路径(26 位 ULID 后缀)。
+    suffix = "01arz3ndektsv4rrffq69g5fav"  # 26 chars
+    assert (
+        _strip_uniqueness_suffix(f"sub/data-{suffix}.csv") == "sub/data.csv"
+    )
+    assert _strip_uniqueness_suffix(f"data-{suffix}.csv") == "data.csv"
+    assert _strip_uniqueness_suffix(f"data-{suffix}") == "data"
+    # 普通文件名不受影响(短后缀、无连字符、目录含连字符)。
+    assert _strip_uniqueness_suffix("sub/data.csv") == "sub/data.csv"
+    assert _strip_uniqueness_suffix("data-v1.csv") == "data-v1.csv"
+    assert _strip_uniqueness_suffix("my-dir/data.csv") == "my-dir/data.csv"
+    assert _strip_uniqueness_suffix("data") == "data"
+
+
 def test_data_resources_model_allows_duplicate_storage_object_reference() -> None:
     """With uk_data_resources_object dropped, no unique index covers
     storage_object_id, so multiple DataResources rows may reference the

common/src/common/db/models/storage.py

@@ -120,9 +120,9 @@ class StorageObjects(Base):
         String(1100),
         comment=(
             "Path inside the trash bucket where the soft-deleted bytes "
-            "live. Format: '{source_bucket}/{object_key}' so a restore "
+            "live. Format: '{source_purpose}/{object_key}-{storage_object_id}' "
-            "is a same-key copy back to the source bucket. NULL while "
+            "(id 后缀防止同名文件多次删除互相覆盖); 恢复时去掉该 id 后缀 "
-            "the row is still available."
+            "拷回原 object_key。NULL while the row is still available."
         ),
     )