diff --git a/backend/AGENTS.md b/backend/AGENTS.md index d95a3690f..abf6ef277 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -1464,7 +1464,7 @@ Multi-file upload with automatic document conversion: - Exact-name generation leases use a portable NFC-plus-casefold coordination key with Win32 trailing-dot/space folding, so filesystem aliases cannot bypass an active generation; filenames that Windows cannot represent losslessly are rejected before staging, while exact legacy POSIX basenames remain deletable after upgrade. Legacy deletion has a POSIX-only lease mode for literal backslashes and components made entirely from dots/spaces; it does not weaken new-upload validation. The original filename remains the published name. Publication tries each candidate lease without blocking and treats a busy canonical key as a collision, so same-batch and inverse concurrent batches advance to a UTF-8-bounded `_N` candidate instead of deadlocking while retaining earlier generations; pathological long suffixes fall back to truncating the complete basename. Deletion rejects an inode that moved outside the requested name's lease. Primary deletion and rollback move the selected primary and its exact generated conversion into one inode-and-intent-tagged transaction directory under the protected conversion namespace before side effects; the primary lives in a dedicated child directory so legal filenames such as `.commit`, `.restore`, and `.conversion` cannot collide with transaction controls. User deletion starts with restore-on-crash intent. Immediately before the first possible remote mutation, a persisted commit marker makes crash recovery discard the authoritative host tombstones; a live, fully compensated remote failure clears it while restoring both host entries. A live rollback writes a restore marker before publishing either hard link, so recovery can finish restoring both entries even when the transaction began with discard intent. A partial remote failure is compensated from the staged authoritative bytes before host restoration; if compensation fails, every remote target is driven back toward the idempotent deletion outcome and host deletion commits instead of publishing a split generation. Failed-upload rollback tombstones otherwise use discard recovery, while legacy primary-only transactions retain their recorded restore/discard behavior. Neither transaction kind is treated as an ordinary disposable upload stage. Final lease release is the commit point: cancellation newly arriving during release is delayed and swallowed so a committed upload is returned as success rather than an indeterminate cancelled result. Embedded-client finalization attempts every lease release and conversion-pool shutdown independently, logging cleanup failures rather than changing an already-committed response or stranding later leases. - Filenames containing NUL, `<`, `>`, or reserved model-context boundary markers are rejected before staging so accepted filenames and exact virtual paths remain lossless in model-visible upload context. Legacy files discovered on disk are still neutralized when listed. - Gateway HTTP uploads use same-directory `.upload-*.part` staging files. Each active stage holds a cross-process liveness lock under `.upload-conversions/.locks/stages/`; startup cleanup skips held stages and sweeps only crash-orphaned files, while `.upload-delete---.part/` transaction directories are resolved according to their persisted crash intent. Legacy intent-less deletion transactions retain restore behavior. Cancellation during staging creation drains the worker and aborts the returned stage before propagating. Staging files are hidden from upload listings, agent upload context, and sandbox listing/search tools. -- Generated Markdown is owned by `user-data/.upload-conversions/.md` and is omitted from primary upload listings. Deletion holds the generation lease while it stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; journal-abort errors cannot skip that host restore. Failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal, commit marker, and exact hidden host tombstones; that retained primary persistently reserves its portable filename coordination key, so a new upload advances to `_N` rather than letting an old path-based retry delete the new remote generation. The v3 journal binds the provider type, backend-issued namespace, immutable sandbox incarnation, and raw sandbox ID. A failed directory fsync after unlink recreates the visible journal before returning, preserving that reservation. Gateway startup and later embedded deletion preparation may use only the provider's non-creating exact reconnect/confirmed-absence result—never `acquire()`, a replacement sandbox for the thread, or a different provider/backend/incarnation—before ordinary transaction cleanup removes the tombstones; an unknown lookup stays pending. AIO local Docker uses daemon/container IDs, while the remote Provisioner uses Kubernetes Namespace/Pod UIDs and treats Pod presence as authoritative even when its Service is gone. Providers without an explicit restart-safe reconciliation identity fail deletion preparation before host staging or remote mutation. Live journals remain protected by their stage lease, reconciliation takes the same name lease used by publication, and a disappearing unrelated transaction is treated as completed after the scanner proves it does not reserve the candidate. Exact legacy POSIX basenames use the existing-file deletion validator during retry. It never guesses or deletes a legacy/user-owned `uploads/.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads. +- Generated Markdown is owned by `user-data/.upload-conversions/.md` and is omitted from primary upload listings. Deletion holds the generation lease while it stages both authoritative host paths, writes a durable remote-deletion journal, persists the remote-mutation phase, removes an explicitly synchronized sandbox primary/conversion, and commits the host transaction. A remote failure is reported and restores the host generation only after every possibly removed remote path has been re-published; journal-abort errors cannot skip that host restore. Failed compensation instead retries idempotent removal of every remote target and commits host deletion. Unconfirmed paths keep the journal, commit marker, and exact hidden host tombstones; that retained primary persistently reserves its portable filename coordination key, so a new upload advances to `_N` rather than letting an old path-based retry delete the new remote generation. The v3 journal binds the provider type, backend-issued namespace, immutable sandbox incarnation, and raw sandbox ID. Journal finalization first persists a `.remote-delete.finalizing` guard; if journal-unlink directory fsync and journal restoration both fail, the guard remains a cross-process basename reservation until startup cleanup fsyncs the transaction directory to make the observed journal absence durable. Gateway startup and later embedded deletion preparation may use only the provider's non-creating exact reconnect/confirmed-absence result—never `acquire()`, a replacement sandbox for the thread, or a different provider/backend/incarnation—before ordinary transaction cleanup removes the tombstones; an unknown lookup stays pending. AIO local Docker uses daemon/container IDs, while the remote Provisioner uses Kubernetes Namespace/Pod UIDs and treats Pod presence as authoritative even when its Service is gone. Providers without an explicit restart-safe reconciliation identity fail deletion preparation before host staging or remote mutation. Live journals remain protected by their stage lease, reconciliation takes the same name lease used by publication, and a disappearing unrelated transaction is treated as completed after the scanner proves it does not reserve the candidate. Exact legacy POSIX basenames use the existing-file deletion validator during retry. It never guesses or deletes a legacy/user-owned `uploads/.md` sibling. Outline extraction opens one descriptor, verifies its `fstat` against the current exclusive regular directory entry, and uses that same descriptor for both outline and preview reads. - Gateway HTTP upload/list/delete handlers offload filesystem work through `deerflow.utils.file_io.run_file_io`, a dedicated ContextVar-preserving file IO executor; only operations that may block waiting for a name lease use the separate lease-wait pool. Work needed by an existing lease holder and non-blocking publication stays on the general pool, so waiters cannot starve conversion, rollback, or release. Cold sandbox-provider construction is also offloaded. Gateway, embedded-client, and IM ingresses share provider-aware publication: mounted providers make the exact host paths sandbox-readable; non-mounted providers acquire the sandbox and synchronize the primary plus generated conversion to their exact virtual paths. Each ingress records attempted remote paths before the write can commit and, on later failure or cancellation, calls the provider-neutral `Sandbox.remove_file()` for those paths before host rollback and lease release; the command fallback requires a per-call unpredictable exact success trailer. Embedded multi-file calls retain every publication and receipt until the whole response is built, then roll back the complete batch on failure. WeChat download publication uses the cancellation-safe async lease adapter, so cancellation drains and rolls back a publication worker that completes late. - Mounted upload paths skip both sandbox acquisition and per-file synchronization. For AIO remote/provisioner deployments this requires an explicit, accurate `sandbox.thread_data_mounts: true` plus a Provisioner advertising a compatible mount contract; every upload/delete/artifact decision waits for any in-flight stale capability probe in a worker thread and uses one atomic snapshot. The same frozen version selects the deterministic sandbox ID, mount set, cache/reuse candidate, create precondition, and response validation; if the peer changes versions during acquisition, the whole attempt restarts with a new ID. The required version is sent in the create JSON, and the Provisioner rejects a mismatch before any Kubernetes read or write, preventing a rolling upgrade from creating an orphan under the prior version's ID. After acquisition, consumers use the immutable contract recorded on that sandbox instance rather than the provider's mutable global capability. Unavailable or legacy results use explicit synchronization and are retried without a Gateway restart. A confirmed legacy peer may create thread sandboxes only for the `default` no-auth user, because its primary user-data mount cannot isolate equal thread IDs across authenticated users; authenticated rolling deployments upgrade the Provisioner first. The Provisioner validates the request before its idempotent fast path, labels Pods with hashed user/thread identity and contract version, stores the exact identity plus a mount-signature annotation, and compares the live Pod specification before reuse. Every remote acquisition path, including active-cache and warm-pool reuse, deliberately replays the idempotent create request instead of trusting GET discovery or a liveness response. Contracts newer than v2 are accepted as forward-compatible, but the response must echo the exact frozen version, user, and thread; legacy responses may omit the version or return exact integer zero, never strings or booleans. The negotiated mount-contract version namespaces deterministic AIO sandbox IDs so a container created under another version cannot satisfy the acquisition; active reuse also replaces its client when validation returns a changed service URL. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity. - Agent receives uploaded file list via `UploadsMiddleware` diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md index 5a70e6d2c..ee6115f9e 100644 --- a/backend/docs/FILE_UPLOAD.md +++ b/backend/docs/FILE_UPLOAD.md @@ -110,9 +110,9 @@ DELETE /api/threads/{thread_id}/uploads/{filename} } ``` -删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；即使持久日志的中止清理失败，本地恢复也会在 `finally` 路径完成。如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认时，受保护的日志、提交标记和精确宿主机 tombstone 会一起保留；隐藏的主文件 tombstone 持续占用该文件名的可移植协调键，因此新的同名上传会得到 `_N` 名称，不会被旧日志的按路径重试误删。日志同时绑定 provider 类型、后端不可变命名空间、sandbox 不可变实例 ID 和原始 sandbox ID；日志删除后的目录同步失败会先重建可见日志再返回错误，进程内不会提前释放旧 basename。Gateway 重启或后续嵌入式删除准备只能通过 provider 的非创建型接口精确重连该旧实例，或由同一后端明确确认该实例已经不存在；不会获取线程当前的新 sandbox，也不会跨 provider、后端或同 ID 的新实例清理。AIO 的本地 Docker 使用 daemon/container ID，远端 Provisioner 使用 Kubernetes Namespace/Pod UID；Provisioner 以 Pod 为权威，即使 Service 已经缺失也不会把仍存活的 Pod 报成 absent。其他 provider 若不能提供同等的 restart-safe 身份能力，删除准备会在宿主机暂存和远端删除之前 fail closed。无法确认时继续保留日志与名称预留，远端确认删除或旧实例确认不存在后才清理 tombstone。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞；扫描期间已经完成收尾的无关删除事务会被视为已完成，而不是让当前上传失败。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除，远端日志重试也沿用同一精确旧名称；新上传仍执行严格的跨平台文件名校验。 +删除 `document.pdf` 时，会先等待该实际文件名当前正在进行的上传、转换或沙箱同步生命周期结束。对于非挂载 provider，系统在同一个 generation lease 内先暂存宿主机主文件与精确转换副本、持久化远端删除日志，再删除沙箱中的对应副本。远端失败且所有可能的副作用都成功补偿时，请求会返回错误并恢复宿主机文件；即使持久日志的中止清理失败，本地恢复也会在 `finally` 路径完成。如果补偿无法完整确认，系统会重试将全部远端目标收敛到删除状态并提交宿主机删除，避免重新暴露一个远端已经缺失的宿主机 generation，此时请求仍返回错误。持续未确认时，受保护的日志、提交标记和精确宿主机 tombstone 会一起保留；隐藏的主文件 tombstone 持续占用该文件名的可移植协调键，因此新的同名上传会得到 `_N` 名称，不会被旧日志的按路径重试误删。日志同时绑定 provider 类型、后端不可变命名空间、sandbox 不可变实例 ID 和原始 sandbox ID；日志删除前会先持久化跨进程 finalization guard，目录同步失败时即使可见日志无法重建，所有 Gateway 进程也不会提前释放旧 basename。Gateway 重启或后续嵌入式删除准备只能通过 provider 的非创建型接口精确重连该旧实例，或由同一后端明确确认该实例已经不存在；不会获取线程当前的新 sandbox，也不会跨 provider、后端或同 ID 的新实例清理。AIO 的本地 Docker 使用 daemon/container ID，远端 Provisioner 使用 Kubernetes Namespace/Pod UID；Provisioner 以 Pod 为权威，即使 Service 已经缺失也不会把仍存活的 Pod 报成 absent。其他 provider 若不能提供同等的 restart-safe 身份能力，删除准备会在宿主机暂存和远端删除之前 fail closed。无法确认时继续保留日志与名称预留，远端确认删除或旧实例确认不存在后才清理 tombstone。随后只删除宿主机上它精确拥有的生成资产和主文件。系统不会推断或删除 `uploads/document.md`；该文件可能是用户独立上传的内容。其他文件名不会被这次等待阻塞；扫描期间已经完成收尾的无关删除事务会被视为已完成，而不是让当前上传失败。在 POSIX 部署上，升级前已经存在且能被列表接口返回的 Windows 非兼容文件名（例如 `CON`、`report?.pdf`、含反斜杠、仅由点/空格组成或末尾带空格的名称）仍可按返回的精确名称删除，远端日志重试也沿用同一精确旧名称；新上传仍执行严格的跨平台文件名校验。 -如果日志 unlink 已经可见但目录 fsync 失败，系统会先重建日志；即使磁盘错误导致重建也失败，本进程仍会保留 fail-closed 的文件名占用并阻止后台事务清理。重启后，文件系统要么恢复旧日志并继续协调，要么确认日志已经消失并安全释放占用。 +如果日志 unlink 已经可见但目录 fsync 失败，系统会先尝试重建日志；即使磁盘错误导致重建也失败，已经落盘的 `.remote-delete.finalizing` 仍会跨进程保留 fail-closed 的文件名占用并阻止后台事务清理。启动清理只有在 guard 仍存在时同步事务目录、确认日志缺失已经持久化，才会删除 guard 并安全释放占用；若崩溃后旧日志恢复，则继续按原 generation 协调。 ## 支持的文档格式 diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index 9278c4d37..f7a37c2f4 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -10,7 +10,6 @@ import os import secrets import shutil import stat -import threading from collections.abc import Callable, Iterator from dataclasses import dataclass from pathlib import Path, PureWindowsPath @@ -33,9 +32,6 @@ from deerflow.utils.thread_id import validate_thread_id logger = logging.getLogger(__name__) -_uncertain_remote_journal_lock = threading.Lock() -_uncertain_remote_journal_reservations: set[tuple[str, str, str]] = set() - UPLOAD_STAGING_PREFIX = ".upload-" UPLOAD_STAGING_SUFFIX = ".part" UPLOAD_DELETION_TRANSACTION_PREFIX = ".upload-delete-" @@ -46,6 +42,7 @@ _UPLOAD_DELETION_CONVERSION_TOMBSTONE = ".conversion" _UPLOAD_DELETION_COMMIT_MARKER = ".commit" _UPLOAD_DELETION_RESTORE_MARKER = ".restore" _UPLOAD_DELETION_REMOTE_JOURNAL = ".remote-delete.json" +_UPLOAD_DELETION_FINALIZE_GUARD = ".remote-delete.finalizing" _WINDOWS_FORBIDDEN_FILENAME_CHARS = frozenset('<>:"|?*') @@ -57,37 +54,6 @@ class RemoteDeletionCompensatedError(RuntimeError): """Remote deletion failed but its side effects were fully compensated.""" -def _retain_uncertain_remote_journal_reservation(journal_path: Path, filename: str) -> None: - """Reserve a remote pathname while journal-unlink durability is unknown.""" - base_dir = journal_path.parent.parent.parent / "uploads" - reservation = ( - os.path.abspath(base_dir), - portable_name_coordination_key(filename), - os.path.abspath(journal_path), - ) - with _uncertain_remote_journal_lock: - _uncertain_remote_journal_reservations.add(reservation) - - -def _release_uncertain_remote_journal_reservation(journal_path: Path) -> None: - journal_key = os.path.abspath(journal_path) - with _uncertain_remote_journal_lock: - matches = {reservation for reservation in _uncertain_remote_journal_reservations if reservation[2] == journal_key} - _uncertain_remote_journal_reservations.difference_update(matches) - - -def _uncertain_remote_journal_reserves_name(base_dir: Path, coordination_key: str) -> bool: - base_key = os.path.abspath(base_dir) - with _uncertain_remote_journal_lock: - return any(reservation_base == base_key and reservation_name == coordination_key for reservation_base, reservation_name, _journal in _uncertain_remote_journal_reservations) - - -def _transaction_has_uncertain_remote_journal(transaction_dir: Path) -> bool: - transaction_key = os.path.abspath(transaction_dir) - with _uncertain_remote_journal_lock: - return any(os.path.dirname(journal_key) == transaction_key for _base, _name, journal_key in _uncertain_remote_journal_reservations) - - @dataclass(slots=True) class StagedUpload: """A complete-or-in-progress upload stored under a hidden temporary name.""" @@ -342,8 +308,6 @@ def _pending_remote_deletion_reserves_name(base_dir: Path, coordination_key: str Callers hold the candidate's name lease while scanning, so reconciliation cannot remove the journal between this check and publication. """ - if _uncertain_remote_journal_reserves_name(base_dir, coordination_key): - return True conversion_dir = base_dir.parent / UPLOAD_CONVERSIONS_DIRNAME try: conversion_dir_stat = os.lstat(conversion_dir) @@ -362,6 +326,7 @@ def _pending_remote_deletion_reserves_name(base_dir: Path, coordination_key: str continue transaction_dir = Path(transaction.path) journal_path = transaction_dir / _UPLOAD_DELETION_REMOTE_JOURNAL + finalize_guard_path = transaction_dir / _UPLOAD_DELETION_FINALIZE_GUARD commit_path = transaction_dir / _UPLOAD_DELETION_COMMIT_MARKER primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME try: @@ -386,18 +351,21 @@ def _pending_remote_deletion_reserves_name(base_dir: Path, coordination_key: str entry = matching_entries[0] try: entry_stat = entry.stat(follow_symlinks=False) - journal_stat = os.lstat(journal_path) commit_stat = os.lstat(commit_path) except FileNotFoundError: # Matching cleanup normally shares this name lease. Be robust - # to legacy cleanup paths: once the durable journal is gone, + # to legacy cleanup paths: once both durable controls are gone, # the old remote operation can no longer target a replacement. + continue + control_stats: list[os.stat_result] = [] + for control_path in (journal_path, finalize_guard_path): try: - os.lstat(journal_path) + control_stats.append(os.lstat(control_path)) except FileNotFoundError: continue - return True - if not stat.S_ISREG(journal_stat.st_mode) or journal_stat.st_nlink != 1 or not stat.S_ISREG(commit_stat.st_mode) or commit_stat.st_nlink != 1: + if not control_stats: + continue + if any(not stat.S_ISREG(control_stat.st_mode) or control_stat.st_nlink != 1 for control_stat in control_stats) or not stat.S_ISREG(commit_stat.st_mode) or commit_stat.st_nlink != 1: raise UnsafeUploadPathError("Unsafe pending remote upload deletion transaction") if not stat.S_ISREG(entry_stat.st_mode) or entry_stat.st_nlink != 1: raise UnsafeUploadPathError("Unsafe pending remote upload deletion tombstone") @@ -877,12 +845,18 @@ def _staged_deletion_remote_journal(staged_path: Path) -> Path: return _deletion_transaction_dir_for_staged_path(staged_path) / _UPLOAD_DELETION_REMOTE_JOURNAL -def _staged_deletion_has_remote_journal(staged_path: Path) -> bool: - try: - os.lstat(_staged_deletion_remote_journal(staged_path)) - except FileNotFoundError: - return False - return True +def _staged_deletion_has_remote_control(staged_path: Path) -> bool: + transaction_dir = _deletion_transaction_dir_for_staged_path(staged_path) + for control_name in ( + _UPLOAD_DELETION_REMOTE_JOURNAL, + _UPLOAD_DELETION_FINALIZE_GUARD, + ): + try: + os.lstat(transaction_dir / control_name) + except FileNotFoundError: + continue + return True + return False def _create_deletion_phase_marker(marker: Path, *, error_message: str) -> None: @@ -992,7 +966,7 @@ def _discard_staged_deletion(staged_path: Path) -> None: _staged_conversion_path(staged_path).unlink(missing_ok=True) staged_path.unlink(missing_ok=True) _clear_staged_deletion_restore(staged_path) - if not _staged_deletion_has_remote_journal(staged_path): + if not _staged_deletion_has_remote_control(staged_path): _clear_staged_deletion_commit(staged_path) _finish_deletion_transaction(staged_path) @@ -1003,7 +977,7 @@ def _finalize_committed_staged_deletion(staged_path: Path) -> None: A remaining journal must retain the exact primary tombstone so publication can reserve its portable filename until reconciliation finishes. """ - if _staged_deletion_has_remote_journal(staged_path): + if _staged_deletion_has_remote_control(staged_path): return _discard_staged_deletion(staged_path) @@ -1043,14 +1017,17 @@ def _finish_deletion_transaction(staged_path: Path) -> None: if not (transaction_dir.name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and transaction_dir.name.endswith(UPLOAD_STAGING_SUFFIX) and transaction_dir.parent.name == UPLOAD_CONVERSIONS_DIRNAME): return primary_dir = transaction_dir / _UPLOAD_DELETION_PRIMARY_DIRNAME - try: - os.lstat(transaction_dir / _UPLOAD_DELETION_REMOTE_JOURNAL) - except FileNotFoundError: - pass - else: + for control_name in ( + _UPLOAD_DELETION_REMOTE_JOURNAL, + _UPLOAD_DELETION_FINALIZE_GUARD, + ): + try: + os.lstat(transaction_dir / control_name) + except FileNotFoundError: + continue # Keep the empty primary container as the on-disk layout discriminator; # older direct-layout transactions may legitimately have a primary - # named exactly like the new journal control. + # named exactly like a current control. return try: primary_dir.rmdir() @@ -1076,8 +1053,6 @@ def _finish_deletion_transaction(staged_path: Path) -> None: def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: """Restore a crash-abandoned tombstone whose basename records its target.""" - if _transaction_has_uncertain_remote_journal(transaction_dir): - return False metadata = _deletion_transaction_metadata(transaction_dir.name) if metadata is None: logger.warning("Refusing malformed upload deletion transaction name: %s", transaction_dir) @@ -1112,6 +1087,7 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: commit_entries = [entry for entry in entries if entry.path not in legacy_primary_paths and entry.name == _UPLOAD_DELETION_COMMIT_MARKER] restore_entries = [entry for entry in entries if entry.path not in legacy_primary_paths and entry.name == _UPLOAD_DELETION_RESTORE_MARKER] remote_journal_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_REMOTE_JOURNAL and primary_dir_entries] + finalize_guard_entries = [entry for entry in entries if entry.name == _UPLOAD_DELETION_FINALIZE_GUARD and primary_dir_entries] legacy_unknown_entries = [ entry for entry in entries @@ -1127,6 +1103,7 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: or len(commit_entries) > 1 or len(restore_entries) > 1 or len(remote_journal_entries) > 1 + or len(finalize_guard_entries) > 1 or len(primary_dir_entries) > 1 or len(legacy_primary_entries) > 1 or legacy_unknown_entries @@ -1143,6 +1120,11 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: if not stat.S_ISREG(remote_journal_stat.st_mode) or remote_journal_stat.st_nlink != 1: logger.warning("Refusing malformed remote upload deletion journal: %s", transaction_dir) return False + if finalize_guard_entries: + finalize_guard_stat = finalize_guard_entries[0].stat(follow_symlinks=False) + if not stat.S_ISREG(finalize_guard_stat.st_mode) or finalize_guard_stat.st_nlink != 1: + logger.warning("Refusing malformed remote upload deletion finalization guard: %s", transaction_dir) + return False primary_dir: Path | None = None if primary_dir_entries: primary_dir_entry = primary_dir_entries[0] @@ -1170,15 +1152,17 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: logger.warning("Refusing malformed upload deletion commit marker: %s", transaction_dir) return False recover_on_crash = False - elif remote_journal_entries: + elif remote_journal_entries or finalize_guard_entries: if not recover_on_crash: - logger.warning("Refusing discard transaction with an uncommitted remote journal: %s", transaction_dir) + logger.warning("Refusing discard transaction with an uncommitted remote control: %s", transaction_dir) return False # The process crashed after preparing the durable remote operation but # before persisting permission to start it. No remote side effect could - # have begun, so discard the prepared journal and restore the host. - _unlink_deletion_control_durably(Path(remote_journal_entries[0].path)) + # have begun, so discard all prepared controls and restore the host. + for control_entry in (*remote_journal_entries, *finalize_guard_entries): + _unlink_deletion_control_durably(Path(control_entry.path)) remote_journal_entries = [] + finalize_guard_entries = [] if restore_entries: restore_stat = restore_entries[0].stat(follow_symlinks=False) if not stat.S_ISREG(restore_stat.st_mode) or restore_stat.st_nlink != 1: @@ -1194,6 +1178,18 @@ def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool: # remote reconciler may clear the journal; a later cleanup pass then # commits these host tombstones. return False + if finalize_guard_entries: + # The journal unlink was visible, but the process did not finish + # removing its persistent guard. Fsync the transaction directory while + # the guard still reserves the basename: this makes the observed journal + # absence durable before any process may publish a replacement. + directory_descriptor = os.open(transaction_dir, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) + _unlink_deletion_control_durably(Path(finalize_guard_entries[0].path)) + finalize_guard_entries = [] if not recover_on_crash and conversion_entries and conversion_stat.st_nlink != 1: logger.warning( "Refusing discard transaction with a linked conversion tombstone: %s", diff --git a/backend/packages/harness/deerflow/uploads/sandbox_sync.py b/backend/packages/harness/deerflow/uploads/sandbox_sync.py index 59ec99479..6e07cb82a 100644 --- a/backend/packages/harness/deerflow/uploads/sandbox_sync.py +++ b/backend/packages/harness/deerflow/uploads/sandbox_sync.py @@ -26,14 +26,14 @@ from deerflow.uploads.layout import conversion_virtual_path from deerflow.uploads.lease import UploadNameLease, UploadStageLease from deerflow.uploads.manager import ( _UPLOAD_DELETION_COMMIT_MARKER, + _UPLOAD_DELETION_FINALIZE_GUARD, _UPLOAD_DELETION_PRIMARY_DIRNAME, RemoteDeletionCommitRequiredError, RemoteDeletionCompensatedError, _deletion_transaction_metadata, _normalize_existing_filename, - _release_uncertain_remote_journal_reservation, - _retain_uncertain_remote_journal_reservation, _staged_deletion_remote_journal, + _unlink_deletion_control_durably, cleanup_stale_upload_staging_files, make_upload_file_sandbox_readable, upload_virtual_path, @@ -74,13 +74,67 @@ def _remove_remote_paths(sandbox: Any, virtual_paths: tuple[str, ...]) -> None: raise first_error +def _fsync_parent_directory(path: Path) -> None: + directory_descriptor = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(directory_descriptor) + finally: + os.close(directory_descriptor) + + +def _write_all(descriptor: int, payload: bytes, *, error_message: str) -> None: + view = memoryview(payload) + while view: + written = os.write(descriptor, view) + if written <= 0: + raise OSError(error_message) + view = view[written:] + + +def _ensure_finalize_guard_durable( + guard_path: Path, + journal_data: dict[str, Any], + journal_payload: bytes, +) -> None: + try: + descriptor = os.open(guard_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + if _read_remote_delete_journal(guard_path) != journal_data: + raise ValueError("remote deletion finalization guard does not match journal") + else: + try: + _write_all( + descriptor, + journal_payload, + error_message="Failed to write remote deletion finalization guard", + ) + os.fsync(descriptor) + except BaseException: + try: + os.close(descriptor) + finally: + try: + _unlink_deletion_control_durably(guard_path) + except BaseException: + logger.error( + "Failed to remove an incomplete remote deletion finalization guard: %s", + guard_path, + exc_info=True, + ) + raise + else: + os.close(descriptor) + # Existing guards may come from an attempt whose directory fsync failed. + # Persist the reservation before making the journal unlink visible. + _fsync_parent_directory(guard_path) + + def _unlink_journal_durably(journal_path: Path) -> None: try: journal_data = _read_remote_delete_journal(journal_path) except FileNotFoundError: - # A prior unlink whose directory fsync failed may still be absent from - # the live namespace yet reappear after a crash. Only a successful - # unlink + parent fsync below may release an in-memory reservation. + # A persistent finalization guard, when present, stays reserved until + # startup cleanup durably confirms this observed journal absence. return filename = journal_data.get("filename") if not isinstance(filename, str): @@ -90,24 +144,17 @@ def _unlink_journal_durably(journal_path: Path) -> None: separators=(",", ":"), sort_keys=True, ).encode("utf-8") - _retain_uncertain_remote_journal_reservation(journal_path, filename) + guard_path = journal_path.with_name(_UPLOAD_DELETION_FINALIZE_GUARD) + _ensure_finalize_guard_durable(guard_path, journal_data, journal_payload) try: journal_path.unlink(missing_ok=True) - directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) - try: - os.fsync(directory_descriptor) - finally: - os.close(directory_descriptor) - _release_uncertain_remote_journal_reservation(journal_path) + _fsync_parent_directory(journal_path) except BaseException: # If unlink became visible but its directory fsync failed, allowing the - # basename reservation to disappear in this process would let a new - # generation reuse the remote path before a crash resurrected the old - # journal. Recreate the same journal entry before propagating. Even if - # the recreation's directory fsync also fails, its visible entry keeps - # publication reserved for the remainder of this process; after a crash - # either a journal survives (reserved) or neither journal does (nothing - # can replay against a later generation). + # basename reservation to disappear would let a new generation reuse + # the remote path before a crash resurrected the old journal. Recreate + # the same journal entry when possible; the already-durable guard keeps + # every process fail-closed even when recreation also fails. try: descriptor = os.open(journal_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) except FileExistsError: @@ -117,12 +164,11 @@ def _unlink_journal_durably(journal_path: Path) -> None: else: restored = False try: - view = memoryview(journal_payload) - while view: - written = os.write(descriptor, view) - if written <= 0: - raise OSError("Failed to restore remote deletion journal") - view = view[written:] + _write_all( + descriptor, + journal_payload, + error_message="Failed to restore remote deletion journal", + ) os.fsync(descriptor) restored = True except BaseException: @@ -131,14 +177,11 @@ def _unlink_journal_durably(journal_path: Path) -> None: os.close(descriptor) if restored: try: - directory_descriptor = os.open(journal_path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) - try: - os.fsync(directory_descriptor) - finally: - os.close(directory_descriptor) + _fsync_parent_directory(journal_path) except BaseException: logger.error("Failed to persist a restored remote deletion journal directory entry: %s", journal_path, exc_info=True) raise + _unlink_deletion_control_durably(guard_path) def _read_remote_delete_journal(journal_path: Path) -> dict[str, Any]: diff --git a/backend/tests/test_upload_sandbox_sync.py b/backend/tests/test_upload_sandbox_sync.py index 7c868cf2a..3b252b712 100644 --- a/backend/tests/test_upload_sandbox_sync.py +++ b/backend/tests/test_upload_sandbox_sync.py @@ -1,7 +1,9 @@ import asyncio import json +import multiprocessing import os import stat +from pathlib import Path from unittest.mock import patch import pytest @@ -34,6 +36,14 @@ def _deletion_hook_for_sandbox(sandbox, **kwargs): return _raw_deletion_hook_for_sandbox(sandbox, **kwargs) +def _publish_upload_in_process(uploads_dir, filename, payload, outcomes): + try: + published = publish_upload_bytes(Path(uploads_dir), filename, payload) + outcomes.put(published.name) + except BaseException as exc: # pragma: no cover - surfaced in the parent + outcomes.put(repr(exc)) + + class _ExactProviderMixin: def reconnect_sandbox_for_reconciliation( self, @@ -220,6 +230,53 @@ def test_remote_delete_finalize_failure_never_restores_deleted_host_generation(t assert cleanup_stale_upload_staging_files(tmp_path) == 0 +def test_finalize_guard_removal_failure_retains_host_tombstone_and_basename(tmp_path): + import deerflow.uploads.sandbox_sync as sandbox_sync_module + + primary_virtual = upload_virtual_path("report.pdf") + + class Sandbox: + def __init__(self) -> None: + self.files = {primary_virtual: b"pdf"} + + def remove_file(self, path: str) -> None: + self.files.pop(path, None) + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"pdf") + sandbox = Sandbox() + deletion_hook = _deletion_hook_for_sandbox( + sandbox, + sandbox_id="sandbox-1", + thread_id="thread-1", + user_id="alice", + provider_key="tests.Provider", + ) + real_unlink_control = sandbox_sync_module._unlink_deletion_control_durably + + def fail_guard_removal(path): + if path.name == ".remote-delete.finalizing": + raise OSError("finalize guard removal failed") + return real_unlink_control(path) + + with patch.object( + sandbox_sync_module, + "_unlink_deletion_control_durably", + side_effect=fail_guard_removal, + ): + with pytest.raises(OSError, match="finalize guard removal failed"): + delete_file_safe(uploads, primary.name, delete_remote_copy=deletion_hook) + + transaction = next((uploads.parent / ".upload-conversions").glob(".upload-delete-*.part")) + assert not (transaction / ".remote-delete.json").exists() + assert (transaction / ".remote-delete.finalizing").is_file() + assert (transaction / "primary" / primary.name).read_bytes() == b"pdf" + assert publish_upload_bytes(uploads, primary.name, b"new").name == "report_1.pdf" + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + + def test_directory_fsync_failure_restores_journal_and_reserves_old_basename(tmp_path): import deerflow.uploads.manager as upload_manager_module import deerflow.uploads.sandbox_sync as sandbox_sync_module @@ -281,18 +338,24 @@ def test_directory_fsync_failure_restores_journal_and_reserves_old_basename(tmp_ real_fsync = sandbox_sync_module.os.fsync fsync_calls = 0 - def fail_first_directory_fsync(descriptor): + def fail_journal_unlink_directory_fsync(descriptor): nonlocal fsync_calls fsync_calls += 1 - if fsync_calls == 1: + if fsync_calls == 3: raise OSError("directory fsync failed") return real_fsync(descriptor) - with patch.object(sandbox_sync_module.os, "fsync", side_effect=fail_first_directory_fsync): + with patch.object( + sandbox_sync_module.os, + "fsync", + side_effect=fail_journal_unlink_directory_fsync, + ): with pytest.raises(OSError, match="directory fsync failed"): sandbox_sync_module._unlink_journal_durably(journal_path) assert journal_path.read_bytes() == journal_payload + finalize_guard = journal_path.with_name(".remote-delete.finalizing") + assert finalize_guard.read_bytes() == journal_payload replacement = publish_upload_bytes(uploads, "report.pdf", b"new primary") assert replacement.name == "report_1.pdf" @@ -306,9 +369,10 @@ def test_directory_fsync_failure_restores_journal_and_reserves_old_basename(tmp_ assert sandbox.files == {} assert replacement.read_bytes() == b"new primary" assert cleanup_stale_upload_staging_files(tmp_path) == 1 + assert not finalize_guard.exists() -def test_failed_journal_restore_keeps_in_memory_name_reservation(tmp_path): +def test_failed_journal_restore_keeps_cross_process_name_reservation(tmp_path): import deerflow.uploads.manager as upload_manager_module import deerflow.uploads.sandbox_sync as sandbox_sync_module @@ -349,7 +413,9 @@ def test_failed_journal_restore_keeps_in_memory_name_reservation(tmp_path): def fail_directory_fsync(descriptor): nonlocal fsync_calls fsync_calls += 1 - if fsync_calls == 1: + # The finalize guard's file and parent directory are persisted first. + # Fail the journal-unlink directory fsync that follows them. + if fsync_calls == 3: raise OSError("directory fsync failed") return real_fsync(descriptor) @@ -358,23 +424,37 @@ def test_failed_journal_restore_keeps_in_memory_name_reservation(tmp_path): raise OSError("journal recreation failed") return real_open(path, flags, mode) + with ( + patch.object(sandbox_sync_module.os, "fsync", side_effect=fail_directory_fsync), + patch.object(sandbox_sync_module.os, "open", side_effect=fail_journal_recreation), + ): + with pytest.raises(OSError, match="directory fsync failed"): + sandbox_sync_module._unlink_journal_durably(journal_path) + + finalize_guard = journal_path.with_name(".remote-delete.finalizing") + assert not journal_path.exists() + assert finalize_guard.is_file() + + context = multiprocessing.get_context("spawn") + outcomes = context.Queue() + worker = context.Process( + target=_publish_upload_in_process, + args=(str(uploads), "report.pdf", b"new primary", outcomes), + ) + worker.start() try: - with ( - patch.object(sandbox_sync_module.os, "fsync", side_effect=fail_directory_fsync), - patch.object(sandbox_sync_module.os, "open", side_effect=fail_journal_recreation), - ): - with pytest.raises(OSError, match="directory fsync failed"): - sandbox_sync_module._unlink_journal_durably(journal_path) - - assert not journal_path.exists() - sandbox_sync_module._unlink_journal_durably(journal_path) - assert cleanup_stale_upload_staging_files(tmp_path) == 0 - replacement = publish_upload_bytes(uploads, "report.pdf", b"new primary") - assert replacement.name == "report_1.pdf" + outcome = outcomes.get(timeout=10) + worker.join(10) finally: - upload_manager_module._release_uncertain_remote_journal_reservation(journal_path) + if worker.is_alive(): + worker.terminate() + worker.join(2) + assert worker.exitcode == 0 + assert outcome == "report_1.pdf" assert cleanup_stale_upload_staging_files(tmp_path) == 1 + assert not finalize_guard.exists() + assert (uploads / "report_1.pdf").read_bytes() == b"new primary" @pytest.mark.parametrize("unlink_first", [False, True], ids=["unlink", "directory-fsync"]) diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 347d0c6d9..da12a412c 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -965,7 +965,15 @@ class TestCleanupStaleUploadStagingFiles: @pytest.mark.parametrize( "filename", - ["report.pdf", "primary", ".remote-delete.json", ".commit", ".restore", ".conversion"], + [ + "report.pdf", + "primary", + ".remote-delete.json", + ".remote-delete.finalizing", + ".commit", + ".restore", + ".conversion", + ], ) def test_restores_legacy_intentless_deletion_transaction(self, tmp_path, filename): import deerflow.uploads.manager as upload_manager_module @@ -1141,7 +1149,17 @@ class TestCleanupStaleUploadStagingFiles: assert conversion.read_text(encoding="utf-8") == "conversion" assert not staged_path.exists() - @pytest.mark.parametrize("filename", [".commit", ".restore", ".conversion", ".remote-delete.json", "primary"]) + @pytest.mark.parametrize( + "filename", + [ + ".commit", + ".restore", + ".conversion", + ".remote-delete.json", + ".remote-delete.finalizing", + "primary", + ], + ) def test_crash_recovery_restores_control_named_primary_and_conversion(self, tmp_path, filename): import deerflow.uploads.manager as upload_manager_module @@ -1193,6 +1211,99 @@ class TestCleanupStaleUploadStagingFiles: assert not conversion.exists() assert not staged_path.exists() + def test_committed_finalize_guard_confirms_journal_absence_before_cleanup(self, tmp_path): + import deerflow.uploads.manager as upload_manager_module + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"primary") + identity = UploadIdentity.from_path(primary) + staged_path, stage_lease = upload_manager_module._stage_primary_deletion( + uploads, + primary, + identity, + recover_on_crash=True, + ) + upload_manager_module._mark_staged_deletion_committed(staged_path) + transaction_dir = staged_path.parent.parent + finalize_guard = transaction_dir / upload_manager_module._UPLOAD_DELETION_FINALIZE_GUARD + finalize_guard.write_text("{}", encoding="utf-8") + stage_lease.release() + + confirmed_before_guard_unlink = False + real_fsync = upload_manager_module.os.fsync + transaction_identity = os.stat(transaction_dir) + + def observe_fsync(descriptor): + nonlocal confirmed_before_guard_unlink + descriptor_stat = os.fstat(descriptor) + if (descriptor_stat.st_dev, descriptor_stat.st_ino) == ( + transaction_identity.st_dev, + transaction_identity.st_ino, + ) and finalize_guard.exists(): + confirmed_before_guard_unlink = True + return real_fsync(descriptor) + + with patch.object(upload_manager_module.os, "fsync", side_effect=observe_fsync): + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + + assert confirmed_before_guard_unlink + assert not transaction_dir.exists() + assert not primary.exists() + + def test_committed_journal_and_finalize_guard_remain_pending(self, tmp_path): + import deerflow.uploads.manager as upload_manager_module + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"primary") + identity = UploadIdentity.from_path(primary) + staged_path, stage_lease = upload_manager_module._stage_primary_deletion( + uploads, + primary, + identity, + recover_on_crash=True, + ) + upload_manager_module._mark_staged_deletion_committed(staged_path) + transaction_dir = staged_path.parent.parent + journal = transaction_dir / upload_manager_module._UPLOAD_DELETION_REMOTE_JOURNAL + finalize_guard = transaction_dir / upload_manager_module._UPLOAD_DELETION_FINALIZE_GUARD + journal.write_text("{}", encoding="utf-8") + finalize_guard.write_text("{}", encoding="utf-8") + stage_lease.release() + + assert cleanup_stale_upload_staging_files(tmp_path) == 0 + assert journal.is_file() + assert finalize_guard.is_file() + assert staged_path.read_bytes() == b"primary" + + def test_uncommitted_journal_and_finalize_guard_restore_host(self, tmp_path): + import deerflow.uploads.manager as upload_manager_module + + uploads = tmp_path / "users" / "alice" / "threads" / "thread-1" / "user-data" / "uploads" + uploads.mkdir(parents=True) + primary = uploads / "report.pdf" + primary.write_bytes(b"primary") + identity = UploadIdentity.from_path(primary) + staged_path, stage_lease = upload_manager_module._stage_primary_deletion( + uploads, + primary, + identity, + recover_on_crash=True, + ) + transaction_dir = staged_path.parent.parent + journal = transaction_dir / upload_manager_module._UPLOAD_DELETION_REMOTE_JOURNAL + finalize_guard = transaction_dir / upload_manager_module._UPLOAD_DELETION_FINALIZE_GUARD + journal.write_text("{}", encoding="utf-8") + finalize_guard.write_text("{}", encoding="utf-8") + stage_lease.release() + + assert cleanup_stale_upload_staging_files(tmp_path) == 1 + assert primary.read_bytes() == b"primary" + assert not transaction_dir.exists() + # --------------------------------------------------------------------------- # delete_file_safe @@ -1385,7 +1496,17 @@ class TestDeleteFileSafe: assert primary.read_bytes() == b"PDF" assert owned_conversion.read_text(encoding="utf-8") == "generated" - @pytest.mark.parametrize("filename", [".commit", ".restore", ".conversion", ".remote-delete.json", "primary"]) + @pytest.mark.parametrize( + "filename", + [ + ".commit", + ".restore", + ".conversion", + ".remote-delete.json", + ".remote-delete.finalizing", + "primary", + ], + ) def test_failed_remote_delete_restores_control_named_primary_and_conversion(self, tmp_path, filename): uploads = tmp_path / "user-data" / "uploads" uploads.mkdir(parents=True)