mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-15 00:19:14 +00:00
fix: recover upload deletes and revalidate remote sandboxes
This commit is contained in:
hetaoBackend
parent
5457406cc0
commit
977be0d445
@ -700,7 +700,7 @@ that cannot tell sibling branches apart.
|
||||
**Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved.
|
||||
**Implementations**:
|
||||
- `LocalSandboxProvider` - Local filesystem execution. `acquire(thread_id)` returns a per-thread `LocalSandbox` (id `local:{thread_id}`) whose `path_mappings` resolve `/mnt/user-data/{workspace,uploads,outputs}` and `/mnt/acp-workspace` to that thread's host directories, so the public `Sandbox` file APIs honour the `/mnt/user-data` contract uniformly with AIO. The more-specific `/mnt/user-data/.upload-conversions` mapping rejects writes through those structured file APIs even though the aggregate `/mnt/user-data` mapping is writable. This is not an OS isolation boundary: explicitly enabled Local host bash operates on host paths outside `PathMapping` write enforcement and must remain disabled for untrusted tasks. `acquire()` / `acquire(None)` keeps the legacy generic singleton (id `local`) for callers without a thread context. Per-thread sandboxes are held in an LRU cache (default 256 entries) guarded by a `threading.Lock`. Public, custom, legacy, and managed integration skill mappings point at stable enabled-only projection roots rather than raw skill directories.
|
||||
- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean requests mounted mode for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Remote mounted mode additionally requires a reachable Provisioner advertising the current mount-contract version; an unavailable probe fails closed, while a confirmed legacy peer may fall back to explicit synchronization only for the `default` no-auth user. Thread mounts create and expose `.upload-conversions` explicitly as read-only, separate from the writable uploads mount. Non-mounted providers omit that nested read-only mount and instead synchronize only the requested generated file into the writable sandbox copy. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support.
|
||||
- `AioSandboxProvider` (`packages/harness/deerflow/community/`) - Docker-based isolation. Active-cache and warm-pool entries are checked with the backend during acquire/reuse; definitively dead containers are dropped from all in-process maps so the thread can discover or create a fresh sandbox instead of reusing a stale client. Remote reuse additionally replays the current idempotent Provisioner create contract (user/thread identity, mounts, and optional runtime/broker flags) before handing out either an active-cache or warm-pool entry. Backend health-check failures are treated as unknown, not dead; local discovery likewise treats an unverifiable container as not adoptable and falls through to create rather than failing acquire. `get()` remains an in-memory lookup for event-loop-safe tool paths — it never touches the ownership store (that would be blocking IO on the event loop); ownership is published on acquire/reclaim and refreshed off the event loop by the dedicated renewal thread (`_renew_owned_leases`). `uses_thread_data_mounts` defaults to backend detection (`LocalContainerBackend=True`, remote/provisioner backends=False), while the optional `sandbox.thread_data_mounts` boolean requests mounted mode for deployments that guarantee the Gateway and sandbox share the same thread user-data directories. Remote mounted mode additionally requires a Provisioner advertising the current mount-contract version; unavailable and confirmed-legacy results use explicit synchronization and are re-probed with bounded backoff/cache expiry, while a non-current peer may create thread sandboxes only for the `default` no-auth user. Thread mounts create and expose `.upload-conversions` explicitly as read-only, separate from the writable uploads mount. Non-mounted providers omit that nested read-only mount and instead synchronize only the requested generated file into the writable sandbox copy. Local-container and hostPath-provisioner mounts use the same stable skill projection roots; PVC-backed skills remain governed by the operator-supplied PVC layout until PVC materialization is implemented. Readiness probes and `agent_sandbox` clients classify loopback/private IPs, single-label cluster hosts, and Docker/Podman internal hostnames as direct control-plane destinations and set `trust_env=False`; external FQDNs and public IPs retain environment proxy support.
|
||||
- `E2BSandboxProvider` (`packages/harness/deerflow/community/e2b_sandbox/`) provides E2B remote isolation.
|
||||
New sandboxes receive a one-shot upload from the enabled-only public, custom,
|
||||
legacy, and managed integration projections. Existing E2B VMs keep their
|
||||
@ -1461,12 +1461,12 @@ Multi-file upload with automatic document conversion:
|
||||
- Reuses one conversion worker per request when called from an active event loop
|
||||
- Files stored in thread-isolated directories under the resolving user's bucket (`users/{user_id}/threads/{thread_id}/user-data/uploads`). For IM channels the owner is threaded explicitly via the `user_id=` kwarg (see IM Channels → Owner-scoped file storage); HTTP/embedded callers resolve it from `get_effective_user_id()`
|
||||
- Every ingress stages a complete payload and atomically publishes it without replacing an existing entry. Collisions across requests, processes, HTTP, embedded, and IM adapters use `name.ext`, `name_1.ext`, `name_2.ext`; storage that cannot provide atomic no-replace publication fails explicitly.
|
||||
- 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 atomically move the selected directory entry into the protected conversion namespace before revalidating its identity, so a pathname replacement between check and removal is restored rather than unlinked. 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.
|
||||
- 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 atomically move the selected directory entry into an inode-tagged transaction directory under the protected conversion namespace before revalidating its identity, so a pathname replacement between check and removal is restored rather than unlinked. Startup recovery restores a crash-abandoned tombstone to its recorded basename, or to a bounded visible recovery name if that basename was recreated; it never treats the tombstone 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. 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.
|
||||
- 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-<inode>-<nonce>.part/` transaction directories are recovered rather than swept. 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/<actual-primary-filename>.md` and is omitted from primary upload listings. Deletion holds the generation lease while it removes an explicitly synchronized sandbox primary/conversion and then the authoritative host paths; a remote failure is reported and leaves the host primary intact. It never guesses or deletes a legacy/user-owned `uploads/<stem>.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 reachable Provisioner advertising the current mount contract; an unavailable capability probe fails closed instead of being treated as confirmed legacy. A confirmed legacy peer may use explicit synchronization 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. Remote acquisition deliberately replays the idempotent create request instead of trusting GET discovery, so an existing Pod must match the complete mounts requested by the current run. `SANDBOX_MOUNT_CONTRACT_VERSION` changes deterministic AIO sandbox IDs so an older container cannot satisfy the new thread acquisition. Reconciliation may still enumerate/adopt the old ID for orphan cleanup; it is not selected for the new identity.
|
||||
- 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 the current mount contract; unavailable or legacy capability 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; a Gateway that negotiated v2 also rejects any create response that does not echo the exact current contract, user, and thread. `SANDBOX_MOUNT_CONTRACT_VERSION` changes deterministic AIO sandbox IDs so an older container cannot satisfy the new thread acquisition. 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`
|
||||
|
||||
See [docs/FILE_UPLOAD.md](docs/FILE_UPLOAD.md) for details.
|
||||
|
||||
@ -277,7 +277,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
try:
|
||||
removed_upload_staging_files = await asyncio.to_thread(cleanup_stale_upload_staging_files)
|
||||
if removed_upload_staging_files:
|
||||
logger.info("Removed %d stale upload staging file(s)", removed_upload_staging_files)
|
||||
logger.info("Cleaned or recovered %d stale upload staging transaction(s)", removed_upload_staging_files)
|
||||
except Exception:
|
||||
logger.warning("Upload staging file cleanup skipped", exc_info=True)
|
||||
|
||||
|
||||
@ -34,11 +34,11 @@ POST /api/threads/{thread_id}/uploads
|
||||
{
|
||||
"filename": "document.pdf",
|
||||
"size": 1234567,
|
||||
"path": ".deer-flow/threads/{thread_id}/user-data/uploads/document.pdf",
|
||||
"path": ".deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/document.pdf",
|
||||
"virtual_path": "/mnt/user-data/uploads/document.pdf",
|
||||
"artifact_url": "/api/threads/{thread_id}/artifacts/mnt/user-data/uploads/document.pdf",
|
||||
"markdown_file": "document.pdf.md",
|
||||
"markdown_path": ".deer-flow/threads/{thread_id}/user-data/.upload-conversions/document.pdf.md",
|
||||
"markdown_path": ".deer-flow/users/{user_id}/threads/{thread_id}/user-data/.upload-conversions/document.pdf.md",
|
||||
"markdown_virtual_path": "/mnt/user-data/.upload-conversions/document.pdf.md",
|
||||
"markdown_artifact_url": "/api/threads/{thread_id}/artifacts/mnt/user-data/.upload-conversions/document.pdf.md"
|
||||
}
|
||||
@ -84,7 +84,7 @@ GET /api/threads/{thread_id}/uploads/list
|
||||
{
|
||||
"filename": "document.pdf",
|
||||
"size": 1234567,
|
||||
"path": ".deer-flow/threads/{thread_id}/user-data/uploads/document.pdf",
|
||||
"path": ".deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/document.pdf",
|
||||
"virtual_path": "/mnt/user-data/uploads/document.pdf",
|
||||
"artifact_url": "/api/threads/{thread_id}/artifacts/mnt/user-data/uploads/document.pdf",
|
||||
"extension": ".pdf",
|
||||
@ -176,12 +176,12 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
|
||||
|
||||
**路径映射关系：**
|
||||
- Agent 使用：`/mnt/user-data/uploads/document.pdf`（虚拟路径）
|
||||
- 实际存储：`backend/.deer-flow/threads/{thread_id}/user-data/uploads/document.pdf`
|
||||
- 实际存储：`backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/document.pdf`
|
||||
- 前端访问：`/api/threads/{thread_id}/artifacts/mnt/user-data/uploads/document.pdf`（HTTP URL）
|
||||
- 转换结果：`/mnt/user-data/.upload-conversions/document.pdf.md`（以上传响应的 `markdown_virtual_path` 为准，不要自行推导）
|
||||
|
||||
上传流程采用“线程目录优先”策略：
|
||||
- 先写入 `backend/.deer-flow/threads/{thread_id}/user-data/uploads/` 作为权威存储
|
||||
上传流程采用“用户/线程目录优先”策略：
|
||||
- 先写入 `backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/` 作为权威存储
|
||||
- 本地沙箱（`sandbox_id=local`）直接使用线程目录内容
|
||||
- AIO 挂载模式把 `/mnt/user-data/.upload-conversions` 单独挂载为只读；Local 的结构化文件 API 通过更具体的只读路径映射执行同一规则，但 Local 宿主机 bash 不属于该边界
|
||||
- Gateway、嵌入式 `DeerFlowClient` 和 IM 通道都会执行同一沙箱可见性步骤：挂载型 provider 调整精确发布路径的读取权限；非挂载 provider 获取沙箱后，把本次主文件及生成转换件精确同步到各自虚拟路径
|
||||
@ -189,7 +189,7 @@ read_file(path="/mnt/user-data/.upload-conversions/document.pdf.md")
|
||||
- 嵌入式 `DeerFlowClient.upload_files()` 以整批为事务边界：后续文件失败会逆序撤销本次调用中此前成功的所有远端副本和宿主 generation
|
||||
- 如果 Gateway 与远端沙箱保证挂载同一份线程 user-data（例如正确对齐的共享 PVC、NFS 或 hostPath），可设置 `sandbox.thread_data_mounts: true`；只有 Provisioner 能通过 `/api/capabilities` 证明当前挂载契约时，上传路由才会跳过 sandbox acquire 和逐文件同步
|
||||
- 新 Gateway 与已确认的旧 Provisioner 混合部署时，只有 `default` 无认证用户可降级为显式同步；旧 Provisioner 的主挂载不包含 `user_id`，认证用户必须先升级 Provisioner，否则创建沙箱会 fail closed
|
||||
- `/api/capabilities` 暂时不可达时，显式请求的挂载模式 fail closed，不会把缺少嵌套只读转换挂载的 Pod 标记为当前契约；远端获取会通过幂等创建重新校验本次请求的完整 Pod 挂载签名，而不是只信任 discovery 响应
|
||||
- `/api/capabilities` 暂时不可达时会选择显式同步并按退避窗口重试能力协商，不会把缺少嵌套只读转换挂载的 Pod 标记为当前契约；在重试确认当前租户隔离契约之前，认证用户的沙箱创建仍会 fail closed。远端获取（包括活跃缓存和暖池复用）会通过幂等创建重新校验本次请求的完整 Pod 挂载签名，而不是只信任 discovery 或健康检查响应
|
||||
- 不确定挂载关系时应省略该配置并保留自动检测。错误地设为 `true` 会导致文件只存在于 Gateway 存储、沙箱内不可见
|
||||
|
||||
## 测试示例
|
||||
@ -247,18 +247,22 @@ print(response.json())
|
||||
## 文件存储结构
|
||||
|
||||
```
|
||||
backend/.deer-flow/threads/
|
||||
└── {thread_id}/
|
||||
└── user-data/
|
||||
├── uploads/
|
||||
│ ├── document.pdf # 用户主文件
|
||||
│ ├── document.md # 用户独立上传，绝不按名称推断归属
|
||||
│ └── presentation.pptx
|
||||
└── .upload-conversions/
|
||||
├── document.pdf.md # document.pdf 的生成结果
|
||||
└── presentation.pptx.md # presentation.pptx 的生成结果
|
||||
backend/.deer-flow/users/
|
||||
└── {user_id}/
|
||||
└── threads/
|
||||
└── {thread_id}/
|
||||
└── user-data/
|
||||
├── uploads/
|
||||
│ ├── document.pdf # 用户主文件
|
||||
│ ├── document.md # 用户独立上传，绝不按名称推断归属
|
||||
│ └── presentation.pptx
|
||||
└── .upload-conversions/
|
||||
├── document.pdf.md # document.pdf 的生成结果
|
||||
└── presentation.pptx.md # presentation.pptx 的生成结果
|
||||
```
|
||||
|
||||
旧版 `backend/.deer-flow/threads/{thread_id}/...` 仅用于兼容和迁移已有数据；新写入不得继续使用该非租户隔离布局。
|
||||
|
||||
## 限制
|
||||
|
||||
- 最大文件大小：100MB（可在 nginx.conf 中配置 `client_max_body_size`）
|
||||
@ -308,7 +312,7 @@ backend/.deer-flow/threads/
|
||||
|
||||
1. 确认 UploadsMiddleware 已在 agent.py 中注册
|
||||
2. 检查 thread_id 是否正确
|
||||
3. 确认文件确实已上传到 `backend/.deer-flow/threads/{thread_id}/user-data/uploads/`
|
||||
3. 确认文件确实已上传到 `backend/.deer-flow/users/{user_id}/threads/{thread_id}/user-data/uploads/`
|
||||
4. 非本地沙箱场景下，确认上传接口没有报错（需要成功完成 sandbox 同步）
|
||||
|
||||
## 开发建议
|
||||
|
||||
@ -270,7 +270,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
if mounted and isinstance(backend, RemoteSandboxBackend):
|
||||
if backend.mount_contract_version < SANDBOX_MOUNT_CONTRACT_VERSION:
|
||||
if not backend.mount_contract_capability_known:
|
||||
raise RuntimeError("Provisioner mount compatibility could not be verified; retry after the Provisioner capability endpoint is reachable")
|
||||
logger.warning("Provisioner mount compatibility could not be verified; using explicit synchronization until a retry confirms the current contract")
|
||||
return False
|
||||
logger.warning(
|
||||
"Configured remote thread-data mounts require Provisioner mount contract v%s; peer advertises v%s, so uploads will use explicit synchronization during the rolling upgrade",
|
||||
SANDBOX_MOUNT_CONTRACT_VERSION,
|
||||
@ -809,6 +810,9 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
|
||||
def _get_extra_mounts(self, thread_id: str | None, *, user_id: str | None = None) -> list[tuple[str, str, bool]]:
|
||||
"""Collect all extra mounts for a sandbox (thread-specific + skills)."""
|
||||
backend = getattr(self, "_backend", None)
|
||||
if isinstance(backend, RemoteSandboxBackend):
|
||||
backend.refresh_capabilities_if_stale()
|
||||
mounts: list[tuple[str, str, bool]] = []
|
||||
|
||||
if thread_id:
|
||||
@ -1372,12 +1376,20 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
)
|
||||
return None
|
||||
|
||||
validated_info = self._revalidate_reused_sandbox_contract(
|
||||
thread_id,
|
||||
existing_id,
|
||||
user_id=effective_user_id,
|
||||
)
|
||||
|
||||
with self._lock:
|
||||
if self._thread_sandboxes.get(key) != existing_id:
|
||||
return None
|
||||
if existing_id not in self._sandboxes:
|
||||
self._thread_sandboxes.pop(key, None)
|
||||
return None
|
||||
if validated_info is not None:
|
||||
self._sandbox_infos[existing_id] = validated_info
|
||||
|
||||
suffix = " (post-lock check)" if post_lock else ""
|
||||
logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}")
|
||||
@ -1417,6 +1429,34 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
return None
|
||||
return existing_id
|
||||
|
||||
def _revalidate_reused_sandbox_contract(
|
||||
self,
|
||||
thread_id: str,
|
||||
sandbox_id: str,
|
||||
*,
|
||||
user_id: str,
|
||||
) -> SandboxInfo | None:
|
||||
"""Replay remote creation so the Provisioner validates current mounts.
|
||||
|
||||
Remote POST is idempotent: an existing Pod is returned only after its
|
||||
live identity, mounts, and optional runtime sidecars match this acquire.
|
||||
Local backends keep their existing health-check-only fast path.
|
||||
"""
|
||||
if getattr(self._backend, "requires_create_validation", False) is not True:
|
||||
return None
|
||||
extra_mounts = self._get_extra_mounts(thread_id, user_id=user_id)
|
||||
info = self._backend.create(
|
||||
thread_id,
|
||||
sandbox_id,
|
||||
extra_mounts=extra_mounts or None,
|
||||
user_id=user_id,
|
||||
provision_lark_cli_runtime=self._lark_integration_active(user_id),
|
||||
provision_lark_cli_broker=self._lark_broker_active(user_id),
|
||||
)
|
||||
if info.sandbox_id != sandbox_id:
|
||||
raise RuntimeError(f"Provisioner validated unexpected sandbox {info.sandbox_id!r}; expected {sandbox_id!r}")
|
||||
return info
|
||||
|
||||
def _reclaim_warm_pool_sandbox(
|
||||
self,
|
||||
thread_id: str | None,
|
||||
@ -1453,6 +1493,12 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
)
|
||||
return None
|
||||
|
||||
validated_info = self._revalidate_reused_sandbox_contract(
|
||||
thread_id,
|
||||
sandbox_id,
|
||||
user_id=effective_user_id,
|
||||
)
|
||||
|
||||
# Publish ownership before the warm → active transition: a raise here must
|
||||
# not leave the sandbox tracked as active but unowned (a peer would see an
|
||||
# orphan and reap it mid-turn). On failure the entry stays warm and this
|
||||
@ -1481,6 +1527,8 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider):
|
||||
return None
|
||||
self._warm_pool_identity.pop(sandbox_id, None)
|
||||
info, _ = warm_item
|
||||
if validated_info is not None:
|
||||
info = validated_info
|
||||
sandbox = AioSandbox(id=sandbox_id, base_url=info.sandbox_url)
|
||||
self._sandboxes[sandbox_id] = sandbox
|
||||
self._sandbox_infos[sandbox_id] = info
|
||||
|
||||
@ -20,6 +20,8 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path, PureWindowsPath
|
||||
|
||||
import requests
|
||||
@ -45,6 +47,9 @@ _PROVISIONER_EXTRA_MOUNT_PATHS = {
|
||||
_UPLOADS_CONTAINER_PATH = "/mnt/user-data/uploads"
|
||||
_UPLOAD_CONVERSIONS_CONTAINER_PATH = "/mnt/user-data/.upload-conversions"
|
||||
_UPLOAD_MOUNT_CONTRACT_VERSION = 2
|
||||
_CAPABILITY_LEGACY_REFRESH_SECONDS = 30.0
|
||||
_CAPABILITY_CURRENT_REFRESH_SECONDS = 300.0
|
||||
_CAPABILITY_RETRY_MAX_SECONDS = 30.0
|
||||
|
||||
_LARK_CLI_RUNTIME_CONTAINER_PATH = "/mnt/integrations/lark-cli/runtime"
|
||||
_LARK_CLI_CONFIG_CONTAINER_PATH = "/mnt/integrations/lark-cli/config"
|
||||
@ -145,6 +150,9 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
self._api_key = api_key
|
||||
self._mount_contract_version = 0
|
||||
self._mount_contract_capability_known = False
|
||||
self._capability_probe_failures = 0
|
||||
self._capability_next_probe_at = time.monotonic() + 1.0
|
||||
self._capability_probe_lock = threading.Lock()
|
||||
|
||||
@property
|
||||
def provisioner_url(self) -> str:
|
||||
@ -170,6 +178,31 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
|
||||
def probe_capabilities(self) -> None:
|
||||
"""Negotiate optional Provisioner capabilities without breaking old peers."""
|
||||
with self._capability_probe_lock:
|
||||
self._probe_capabilities_locked()
|
||||
|
||||
def refresh_capabilities_if_stale(self) -> bool:
|
||||
"""Refresh a legacy/unavailable capability result after its retry window.
|
||||
|
||||
Returns ``True`` when this call performed the probe. The non-blocking
|
||||
lock prevents a burst of concurrent acquires from serially repeating the
|
||||
same network request after one cached result expires.
|
||||
"""
|
||||
now = time.monotonic()
|
||||
if now < self._capability_next_probe_at:
|
||||
return False
|
||||
if not self._capability_probe_lock.acquire(blocking=False):
|
||||
return False
|
||||
try:
|
||||
if time.monotonic() < self._capability_next_probe_at:
|
||||
return False
|
||||
self._probe_capabilities_locked()
|
||||
return True
|
||||
finally:
|
||||
self._capability_probe_lock.release()
|
||||
|
||||
def _probe_capabilities_locked(self) -> None:
|
||||
"""Probe while ``_capability_probe_lock`` is held."""
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self._provisioner_url}/api/capabilities",
|
||||
@ -179,15 +212,26 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
if getattr(response, "status_code", None) == 404:
|
||||
self._mount_contract_version = 0
|
||||
self._mount_contract_capability_known = True
|
||||
self._capability_probe_failures = 0
|
||||
self._capability_next_probe_at = time.monotonic() + _CAPABILITY_LEGACY_REFRESH_SECONDS
|
||||
return
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
version = payload.get("mount_contract_version", 0) if isinstance(payload, dict) else 0
|
||||
self._mount_contract_version = version if isinstance(version, int) and version >= 0 else 0
|
||||
self._mount_contract_capability_known = True
|
||||
self._capability_probe_failures = 0
|
||||
refresh_seconds = _CAPABILITY_CURRENT_REFRESH_SECONDS if self._mount_contract_version >= _UPLOAD_MOUNT_CONTRACT_VERSION else _CAPABILITY_LEGACY_REFRESH_SECONDS
|
||||
self._capability_next_probe_at = time.monotonic() + refresh_seconds
|
||||
except (requests.RequestException, ValueError, TypeError):
|
||||
self._mount_contract_version = 0
|
||||
self._mount_contract_capability_known = False
|
||||
self._capability_probe_failures += 1
|
||||
retry_seconds = min(
|
||||
_CAPABILITY_RETRY_MAX_SECONDS,
|
||||
float(2 ** (self._capability_probe_failures - 1)),
|
||||
)
|
||||
self._capability_next_probe_at = time.monotonic() + retry_seconds
|
||||
logger.warning("Provisioner mount capabilities are unavailable; mounted thread data will fail closed until compatibility can be verified")
|
||||
|
||||
# ── SandboxBackend interface ──────────────────────────────────────────
|
||||
@ -207,6 +251,8 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
Calls ``POST /api/sandboxes`` which creates a dedicated Pod +
|
||||
NodePort Service in k3s.
|
||||
"""
|
||||
if self._mount_contract_version < _UPLOAD_MOUNT_CONTRACT_VERSION:
|
||||
self.refresh_capabilities_if_stale()
|
||||
effective_user_id = user_id or get_effective_user_id()
|
||||
if self._mount_contract_version < _UPLOAD_MOUNT_CONTRACT_VERSION and thread_id is not None and effective_user_id != DEFAULT_USER_ID:
|
||||
raise RuntimeError(f"Provisioner mount contract v{self._mount_contract_version} cannot isolate user {effective_user_id!r}; upgrade the Provisioner before creating authenticated-user sandboxes")
|
||||
@ -318,10 +364,31 @@ class RemoteSandboxBackend(SandboxBackend):
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if not isinstance(data, dict):
|
||||
raise RuntimeError("Provisioner mount contract response is not an object")
|
||||
effective_thread_id = thread_id or sandbox_id
|
||||
if self._mount_contract_version >= _UPLOAD_MOUNT_CONTRACT_VERSION:
|
||||
response_contract = (
|
||||
data.get("sandbox_id"),
|
||||
data.get("user_id"),
|
||||
data.get("thread_id"),
|
||||
data.get("mount_contract_version"),
|
||||
)
|
||||
expected_contract = (
|
||||
sandbox_id,
|
||||
effective_user_id,
|
||||
effective_thread_id,
|
||||
_UPLOAD_MOUNT_CONTRACT_VERSION,
|
||||
)
|
||||
if response_contract != expected_contract:
|
||||
raise RuntimeError(f"Provisioner mount contract response does not match the requested sandbox identity: expected={expected_contract!r}, received={response_contract!r}")
|
||||
logger.info(f"Provisioner created sandbox {sandbox_id}: sandbox_url={data['sandbox_url']}")
|
||||
return SandboxInfo(
|
||||
sandbox_id=sandbox_id,
|
||||
sandbox_url=data["sandbox_url"],
|
||||
user_id=data.get("user_id"),
|
||||
thread_id=data.get("thread_id"),
|
||||
mount_contract_version=data.get("mount_contract_version"),
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
logger.error(f"Provisioner create failed for {sandbox_id}: {exc}")
|
||||
|
||||
@ -33,6 +33,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
UPLOAD_STAGING_PREFIX = ".upload-"
|
||||
UPLOAD_STAGING_SUFFIX = ".part"
|
||||
UPLOAD_DELETION_TRANSACTION_PREFIX = ".upload-delete-"
|
||||
_WINDOWS_FORBIDDEN_FILENAME_CHARS = frozenset('<>:"|?*')
|
||||
|
||||
|
||||
@ -540,7 +541,7 @@ def _iter_upload_storage_dirs(base_dir: Path):
|
||||
|
||||
|
||||
def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> int:
|
||||
"""Remove orphaned Gateway upload staging files left by a hard crash."""
|
||||
"""Clean upload stages and recover interrupted primary deletions."""
|
||||
root = Path(base_dir) if base_dir is not None else get_paths().base_dir
|
||||
removed = 0
|
||||
for uploads_dir in _iter_upload_storage_dirs(root):
|
||||
@ -553,6 +554,16 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> in
|
||||
try:
|
||||
with os.scandir(uploads_dir) as entries:
|
||||
for entry in entries:
|
||||
if entry.name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and entry.name.endswith(UPLOAD_STAGING_SUFFIX) and entry.is_dir(follow_symlinks=False):
|
||||
stage_lease = UploadStageLease.try_acquire(uploads_dir, entry.name)
|
||||
if stage_lease is None:
|
||||
continue
|
||||
try:
|
||||
if _recover_stale_deletion_transaction(Path(entry.path)):
|
||||
removed += 1
|
||||
finally:
|
||||
stage_lease.release()
|
||||
continue
|
||||
if not is_upload_staging_file(entry.name) or not entry.is_file(follow_symlinks=False):
|
||||
continue
|
||||
stage_lease = UploadStageLease.try_acquire(uploads_dir, entry.name)
|
||||
@ -574,6 +585,17 @@ def cleanup_stale_upload_staging_files(base_dir: Path | str | None = None) -> in
|
||||
return removed
|
||||
|
||||
|
||||
def _deletion_transaction_inode(transaction_name: str) -> int | None:
|
||||
"""Decode the selected generation inode embedded in a transaction name."""
|
||||
if not (transaction_name.startswith(UPLOAD_DELETION_TRANSACTION_PREFIX) and transaction_name.endswith(UPLOAD_STAGING_SUFFIX)):
|
||||
return None
|
||||
body = transaction_name[len(UPLOAD_DELETION_TRANSACTION_PREFIX) : -len(UPLOAD_STAGING_SUFFIX)]
|
||||
inode_hex, separator, token = body.partition("-")
|
||||
if not separator or not inode_hex or len(token) != 32 or any(character not in "0123456789abcdef" for character in inode_hex + token):
|
||||
return None
|
||||
return int(inode_hex, 16)
|
||||
|
||||
|
||||
def write_upload_file_no_symlink(base_dir: Path, filename: str, data: bytes) -> Path:
|
||||
"""Compatibility wrapper for collision-safe upload publication."""
|
||||
return publish_upload_bytes(base_dir, filename, data)
|
||||
@ -613,7 +635,7 @@ def list_files_in_dir(directory: Path) -> dict:
|
||||
return {"files": files, "count": len(files)}
|
||||
|
||||
|
||||
def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> Path:
|
||||
def _scan_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> Path:
|
||||
"""Return the directory entry that actually names *identity*."""
|
||||
matching_path: Path | None = None
|
||||
with os.scandir(base_dir) as entries:
|
||||
@ -637,6 +659,11 @@ def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> P
|
||||
return matching_path
|
||||
|
||||
|
||||
def _find_upload_path_by_identity(base_dir: Path, identity: UploadIdentity) -> Path:
|
||||
"""Re-scan for the selected identity immediately before deletion staging."""
|
||||
return _scan_upload_path_by_identity(base_dir, identity)
|
||||
|
||||
|
||||
def _restore_staged_deletion(
|
||||
staged_path: Path,
|
||||
original_path: Path,
|
||||
@ -647,10 +674,7 @@ def _restore_staged_deletion(
|
||||
staged_stat = os.lstat(staged_path)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
if not stat.S_ISREG(staged_stat.st_mode) or (
|
||||
staged_stat.st_dev,
|
||||
staged_stat.st_ino,
|
||||
) != (identity.device, identity.inode):
|
||||
if not stat.S_ISREG(staged_stat.st_mode) or staged_stat.st_nlink != 1 or (staged_stat.st_dev, staged_stat.st_ino) != (identity.device, identity.inode):
|
||||
raise UnsafeUploadPathError("Staged upload deletion changed identity")
|
||||
try:
|
||||
original_stat = os.lstat(original_path)
|
||||
@ -662,35 +686,179 @@ def _restore_staged_deletion(
|
||||
original_stat.st_ino,
|
||||
) == (identity.device, identity.inode):
|
||||
staged_path.unlink()
|
||||
_finish_deletion_transaction(staged_path)
|
||||
return
|
||||
raise UnsafeUploadPathError("Upload name was recreated while deletion was being rolled back")
|
||||
recovery_path = _preserve_staged_entry_as_recovery(staged_path, original_path)
|
||||
logger.warning(
|
||||
"Upload name was recreated during deletion rollback; preserved the prior generation as %s",
|
||||
recovery_path,
|
||||
)
|
||||
_finish_deletion_transaction(staged_path)
|
||||
return
|
||||
try:
|
||||
os.link(staged_path, original_path, follow_symlinks=False)
|
||||
except FileExistsError as exc:
|
||||
raise UnsafeUploadPathError("Upload name was recreated while deletion was being rolled back") from exc
|
||||
except FileExistsError:
|
||||
recovery_path = _preserve_staged_entry_as_recovery(staged_path, original_path)
|
||||
logger.warning(
|
||||
"Upload name was recreated during deletion rollback; preserved the prior generation as %s",
|
||||
recovery_path,
|
||||
)
|
||||
_finish_deletion_transaction(staged_path)
|
||||
return
|
||||
staged_path.unlink()
|
||||
_finish_deletion_transaction(staged_path)
|
||||
|
||||
|
||||
def _recovery_path_for(original_path: Path) -> Path:
|
||||
"""Return a collision-resistant visible recovery name within NAME_MAX."""
|
||||
marker = f"_recovered_{secrets.token_hex(8)}"
|
||||
prefix = _truncate_utf8(
|
||||
original_path.name,
|
||||
255 - len(marker.encode("utf-8")),
|
||||
)
|
||||
return original_path.with_name(f"{prefix}{marker}")
|
||||
|
||||
|
||||
def _preserve_staged_entry_as_recovery(staged_path: Path, original_path: Path) -> Path:
|
||||
"""Publish staged bytes under a visible no-replace recovery name."""
|
||||
while True:
|
||||
recovery_path = _recovery_path_for(original_path)
|
||||
try:
|
||||
os.link(staged_path, recovery_path, follow_symlinks=False)
|
||||
break
|
||||
except FileExistsError:
|
||||
continue
|
||||
staged_path.unlink()
|
||||
return recovery_path
|
||||
|
||||
|
||||
def _finish_deletion_transaction(staged_path: Path) -> None:
|
||||
"""Remove the now-empty transaction directory, when this is the new layout."""
|
||||
transaction_dir = staged_path.parent
|
||||
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
|
||||
try:
|
||||
transaction_dir.rmdir()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except OSError:
|
||||
logger.warning(
|
||||
"Failed to remove completed upload deletion transaction: %s",
|
||||
transaction_dir,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
|
||||
def _recover_stale_deletion_transaction(transaction_dir: Path) -> bool:
|
||||
"""Restore a crash-abandoned tombstone whose basename records its target."""
|
||||
expected_inode = _deletion_transaction_inode(transaction_dir.name)
|
||||
if expected_inode is None:
|
||||
logger.warning("Refusing malformed upload deletion transaction name: %s", transaction_dir)
|
||||
return False
|
||||
try:
|
||||
transaction_stat = os.lstat(transaction_dir)
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
if stat.S_ISLNK(transaction_stat.st_mode) or not stat.S_ISDIR(transaction_stat.st_mode):
|
||||
return False
|
||||
|
||||
entries = list(os.scandir(transaction_dir))
|
||||
if not entries:
|
||||
transaction_dir.rmdir()
|
||||
return True
|
||||
if len(entries) != 1:
|
||||
logger.warning(
|
||||
"Refusing malformed upload deletion transaction with %s entries: %s",
|
||||
len(entries),
|
||||
transaction_dir,
|
||||
)
|
||||
return False
|
||||
entry = entries[0]
|
||||
if not entry.is_file(follow_symlinks=False):
|
||||
logger.warning("Refusing malformed upload deletion transaction: %s", transaction_dir)
|
||||
return False
|
||||
try:
|
||||
original_name = _normalize_existing_filename(entry.name)
|
||||
except ValueError:
|
||||
logger.warning("Refusing upload deletion transaction with unsafe target: %s", transaction_dir)
|
||||
return False
|
||||
if original_name != entry.name:
|
||||
return False
|
||||
|
||||
uploads_dir = transaction_dir.parent.parent / "uploads"
|
||||
name_lease = UploadNameLease.try_acquire(
|
||||
uploads_dir,
|
||||
original_name,
|
||||
allow_legacy_posix_filename=True,
|
||||
)
|
||||
if name_lease is None:
|
||||
return False
|
||||
try:
|
||||
staged_path = Path(entry.path)
|
||||
staged_stat = os.lstat(staged_path)
|
||||
if not stat.S_ISREG(staged_stat.st_mode) or staged_stat.st_nlink not in {1, 2} or staged_stat.st_ino != expected_inode:
|
||||
logger.warning("Refusing unsafe upload deletion tombstone: %s", staged_path)
|
||||
return False
|
||||
identity = UploadIdentity(
|
||||
device=staged_stat.st_dev,
|
||||
inode=staged_stat.st_ino,
|
||||
)
|
||||
if staged_stat.st_nlink == 2:
|
||||
# A previous recovery may have crashed after publishing the visible
|
||||
# hard link but before removing the tombstone. Accept only one
|
||||
# exact visible peer; any additional/hidden alias is ambiguous.
|
||||
visible_matches: list[Path] = []
|
||||
with os.scandir(uploads_dir) as upload_entries:
|
||||
for upload_entry in upload_entries:
|
||||
try:
|
||||
upload_stat = upload_entry.stat(follow_symlinks=False)
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
if stat.S_ISREG(upload_stat.st_mode) and (
|
||||
upload_stat.st_dev,
|
||||
upload_stat.st_ino,
|
||||
) == (identity.device, identity.inode):
|
||||
visible_matches.append(Path(upload_entry.path))
|
||||
if len(visible_matches) != 1:
|
||||
logger.warning(
|
||||
"Refusing ambiguous upload deletion recovery with %s visible aliases: %s",
|
||||
len(visible_matches),
|
||||
staged_path,
|
||||
)
|
||||
return False
|
||||
staged_path.unlink()
|
||||
_finish_deletion_transaction(staged_path)
|
||||
return True
|
||||
_restore_staged_deletion(
|
||||
staged_path,
|
||||
uploads_dir / original_name,
|
||||
identity,
|
||||
)
|
||||
return True
|
||||
finally:
|
||||
name_lease.release()
|
||||
|
||||
|
||||
def _restore_unexpected_staged_entry(staged_path: Path, original_path: Path) -> None:
|
||||
"""Restore an entry moved during an identity race without unlinking it."""
|
||||
staged_stat = os.lstat(staged_path)
|
||||
if not stat.S_ISREG(staged_stat.st_mode) or staged_stat.st_nlink != 1:
|
||||
raise UnsafeUploadPathError("Unsafe staged upload entry cannot be restored")
|
||||
try:
|
||||
os.link(staged_path, original_path, follow_symlinks=False)
|
||||
except FileExistsError:
|
||||
# A non-cooperating writer recreated the original name while the entry
|
||||
# was staged. Preserve the moved payload under a collision-resistant
|
||||
# visible recovery name instead of deleting either generation.
|
||||
while True:
|
||||
recovery_path = original_path.with_name(f"{original_path.stem}_recovered_{secrets.token_hex(8)}{original_path.suffix}")
|
||||
try:
|
||||
os.link(staged_path, recovery_path, follow_symlinks=False)
|
||||
break
|
||||
except FileExistsError:
|
||||
continue
|
||||
recovery_path = _preserve_staged_entry_as_recovery(staged_path, original_path)
|
||||
logger.warning(
|
||||
"Upload entry changed during deletion; preserved the raced entry as %s",
|
||||
recovery_path,
|
||||
)
|
||||
_finish_deletion_transaction(staged_path)
|
||||
return
|
||||
staged_path.unlink()
|
||||
_finish_deletion_transaction(staged_path)
|
||||
|
||||
|
||||
def _stage_primary_deletion(
|
||||
@ -701,19 +869,19 @@ def _stage_primary_deletion(
|
||||
"""Atomically move the selected entry into the protected conversion namespace."""
|
||||
staging_dir = ensure_conversion_dir(base_dir)
|
||||
while True:
|
||||
staged_path = staging_dir / (f"{UPLOAD_STAGING_PREFIX}delete-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
|
||||
stage_lease = UploadStageLease.acquire(staging_dir, staged_path.name)
|
||||
transaction_dir = staging_dir / (f"{UPLOAD_DELETION_TRANSACTION_PREFIX}{identity.inode:x}-{secrets.token_hex(16)}{UPLOAD_STAGING_SUFFIX}")
|
||||
stage_lease = UploadStageLease.acquire(staging_dir, transaction_dir.name)
|
||||
try:
|
||||
os.lstat(staged_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
else:
|
||||
transaction_dir.mkdir(mode=0o700)
|
||||
except FileExistsError:
|
||||
stage_lease.release()
|
||||
continue
|
||||
staged_path = transaction_dir / primary_path.name
|
||||
|
||||
try:
|
||||
os.rename(primary_path, staged_path)
|
||||
except BaseException:
|
||||
transaction_dir.rmdir()
|
||||
stage_lease.release()
|
||||
raise
|
||||
|
||||
@ -795,10 +963,21 @@ def delete_file_safe(
|
||||
raise UnsafeUploadPathError(f"Unsafe upload file: {safe_name}")
|
||||
|
||||
identity = UploadIdentity(device=file_stat.st_dev, inode=file_stat.st_ino)
|
||||
# Capture the exact directory-entry spelling before the final race
|
||||
# check. On case-insensitive or normalization-insensitive filesystems
|
||||
# ``safe_name`` may be only an alias, while generated companions are
|
||||
# keyed by the real published spelling. A second scan below detects a
|
||||
# non-cooperating rename; its new spelling must not redirect companion
|
||||
# cleanup away from this initial generation.
|
||||
initial_file_path = _scan_upload_path_by_identity(base_dir, identity)
|
||||
if portable_name_coordination_key(initial_file_path.name) != portable_name_coordination_key(safe_name):
|
||||
raise UnsafeUploadPathError("Upload name changed outside the requested generation lease")
|
||||
owned_conversion = existing_conversion_path_for_upload(initial_file_path)
|
||||
companion_name = initial_file_path.name
|
||||
|
||||
actual_file_path = _find_upload_path_by_identity(base_dir, identity)
|
||||
if portable_name_coordination_key(actual_file_path.name) != portable_name_coordination_key(safe_name):
|
||||
raise UnsafeUploadPathError("Upload name changed outside the requested generation lease")
|
||||
owned_conversion = existing_conversion_path_for_upload(actual_file_path)
|
||||
final_stat = os.lstat(actual_file_path)
|
||||
if not stat.S_ISREG(final_stat.st_mode) or final_stat.st_nlink != 1 or (final_stat.st_dev, final_stat.st_ino) != (identity.device, identity.inode):
|
||||
raise UnsafeUploadPathError("Upload is no longer an exclusive directory entry")
|
||||
@ -809,10 +988,11 @@ def delete_file_safe(
|
||||
)
|
||||
try:
|
||||
if delete_remote_copy is not None:
|
||||
delete_remote_copy(actual_file_path.name)
|
||||
delete_remote_copy(companion_name)
|
||||
if owned_conversion is not None:
|
||||
owned_conversion.unlink(missing_ok=True)
|
||||
staged_path.unlink()
|
||||
_finish_deletion_transaction(staged_path)
|
||||
except BaseException:
|
||||
_restore_staged_deletion(staged_path, actual_file_path, identity)
|
||||
raise
|
||||
|
||||
@ -429,15 +429,86 @@ def test_remote_backend_distinguishes_unavailable_capability_probe(monkeypatch):
|
||||
assert backend.mount_contract_capability_known is False
|
||||
|
||||
|
||||
def test_remote_mount_override_fails_closed_when_capability_probe_is_unavailable():
|
||||
def test_remote_mount_override_falls_back_to_explicit_sync_when_capability_probe_is_unavailable():
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
|
||||
provider = aio_mod.AioSandboxProvider.__new__(aio_mod.AioSandboxProvider)
|
||||
provider._config = {"thread_data_mounts": True}
|
||||
provider._backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
|
||||
|
||||
with pytest.raises(RuntimeError, match="could not be verified"):
|
||||
_ = provider.uses_thread_data_mounts
|
||||
assert provider.uses_thread_data_mounts is False
|
||||
|
||||
|
||||
def test_remote_backend_retries_unavailable_capability_probe(monkeypatch):
|
||||
remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
|
||||
now = [100.0]
|
||||
responses = [
|
||||
remote_mod.requests.ConnectionError("not ready"),
|
||||
{"mount_contract_version": 2},
|
||||
]
|
||||
|
||||
class _Response:
|
||||
status_code = 200
|
||||
|
||||
def __init__(self, payload):
|
||||
self._payload = payload
|
||||
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
def _get(*_args, **_kwargs):
|
||||
value = responses.pop(0)
|
||||
if isinstance(value, BaseException):
|
||||
raise value
|
||||
return _Response(value)
|
||||
|
||||
monkeypatch.setattr(remote_mod.time, "monotonic", lambda: now[0])
|
||||
monkeypatch.setattr(remote_mod.requests, "get", _get)
|
||||
backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
|
||||
|
||||
backend.probe_capabilities()
|
||||
assert backend.mount_contract_capability_known is False
|
||||
|
||||
now[0] += 1.0
|
||||
backend.refresh_capabilities_if_stale()
|
||||
|
||||
assert backend.mount_contract_capability_known is True
|
||||
assert backend.mount_contract_version == 2
|
||||
|
||||
|
||||
def test_remote_backend_retries_known_legacy_capability_probe(monkeypatch):
|
||||
remote_mod = importlib.import_module("deerflow.community.aio_sandbox.remote_backend")
|
||||
now = [200.0]
|
||||
|
||||
class _Response:
|
||||
def __init__(self, status_code, payload=None):
|
||||
self.status_code = status_code
|
||||
self._payload = payload or {}
|
||||
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
responses = [
|
||||
_Response(404),
|
||||
_Response(200, {"mount_contract_version": 2}),
|
||||
]
|
||||
monkeypatch.setattr(remote_mod.time, "monotonic", lambda: now[0])
|
||||
monkeypatch.setattr(remote_mod.requests, "get", lambda *_args, **_kwargs: responses.pop(0))
|
||||
backend = remote_mod.RemoteSandboxBackend("http://provisioner:8002")
|
||||
|
||||
backend.probe_capabilities()
|
||||
assert backend.mount_contract_version == 0
|
||||
|
||||
now[0] += 30.0
|
||||
backend.refresh_capabilities_if_stale()
|
||||
|
||||
assert backend.mount_contract_version == 2
|
||||
|
||||
|
||||
def test_remote_discovery_carries_verified_identity(monkeypatch):
|
||||
@ -614,6 +685,130 @@ async def test_remote_contract_backend_revalidates_existing_pod_via_create_async
|
||||
assert create_calls == [("thread-1", "sandbox-v2", "alice")]
|
||||
|
||||
|
||||
def _install_remote_revalidation_backend(provider, aio_mod, calls):
|
||||
def create(
|
||||
thread_id,
|
||||
sandbox_id,
|
||||
*,
|
||||
extra_mounts=None,
|
||||
user_id=None,
|
||||
provision_lark_cli_runtime=False,
|
||||
provision_lark_cli_broker=False,
|
||||
):
|
||||
calls.append(
|
||||
{
|
||||
"thread_id": thread_id,
|
||||
"sandbox_id": sandbox_id,
|
||||
"extra_mounts": extra_mounts,
|
||||
"user_id": user_id,
|
||||
"provision_lark_cli_runtime": provision_lark_cli_runtime,
|
||||
"provision_lark_cli_broker": provision_lark_cli_broker,
|
||||
}
|
||||
)
|
||||
return aio_mod.SandboxInfo(
|
||||
sandbox_id=sandbox_id,
|
||||
sandbox_url="http://sandbox-host",
|
||||
user_id=user_id,
|
||||
thread_id=thread_id,
|
||||
mount_contract_version=2,
|
||||
)
|
||||
|
||||
provider._backend = SimpleNamespace(
|
||||
requires_create_validation=True,
|
||||
is_alive=MagicMock(return_value=True),
|
||||
create=create,
|
||||
)
|
||||
provider._ensure_skills_projection = lambda _user_id: None
|
||||
provider._get_extra_mounts = lambda *_args, **_kwargs: [("/host/current", "/mnt/current", True)]
|
||||
provider._lark_integration_active = lambda _user_id: True
|
||||
provider._lark_broker_active = lambda _user_id: False
|
||||
|
||||
|
||||
def test_remote_active_cache_revalidates_current_contract_before_reuse(tmp_path):
|
||||
provider, _sandbox, aio_mod = _make_provider_with_active_sandbox(tmp_path, "sandbox-active")
|
||||
provider._thread_sandboxes = {("alice", "thread-1"): "sandbox-active"}
|
||||
calls: list[dict] = []
|
||||
_install_remote_revalidation_backend(provider, aio_mod, calls)
|
||||
|
||||
result = provider._acquire_internal("thread-1", user_id="alice")
|
||||
|
||||
assert result == "sandbox-active"
|
||||
assert calls == [
|
||||
{
|
||||
"thread_id": "thread-1",
|
||||
"sandbox_id": "sandbox-active",
|
||||
"extra_mounts": [("/host/current", "/mnt/current", True)],
|
||||
"user_id": "alice",
|
||||
"provision_lark_cli_runtime": True,
|
||||
"provision_lark_cli_broker": False,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_remote_active_cache_revalidates_current_contract_before_reuse_async(tmp_path):
|
||||
provider, _sandbox, aio_mod = _make_provider_with_active_sandbox(tmp_path, "sandbox-active-async")
|
||||
provider._thread_sandboxes = {("alice", "thread-1"): "sandbox-active-async"}
|
||||
calls: list[dict] = []
|
||||
_install_remote_revalidation_backend(provider, aio_mod, calls)
|
||||
|
||||
result = await provider._acquire_internal_async("thread-1", user_id="alice")
|
||||
|
||||
assert result == "sandbox-active-async"
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["extra_mounts"] == [("/host/current", "/mnt/current", True)]
|
||||
|
||||
|
||||
def _make_remote_warm_provider(tmp_path, monkeypatch, sandbox_id):
|
||||
aio_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider")
|
||||
provider = _make_provider(tmp_path)
|
||||
provider._lock = aio_mod.threading.Lock()
|
||||
provider._sandboxes = {}
|
||||
provider._sandbox_infos = {}
|
||||
provider._thread_sandboxes = {}
|
||||
provider._last_activity = {}
|
||||
provider._warm_pool = {
|
||||
sandbox_id: (
|
||||
aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox-host"),
|
||||
123.0,
|
||||
)
|
||||
}
|
||||
provider._warm_pool_identity = {sandbox_id: None}
|
||||
provider._sandbox_id_for_thread = lambda *_args, **_kwargs: sandbox_id
|
||||
monkeypatch.setattr(
|
||||
aio_mod,
|
||||
"AioSandbox",
|
||||
lambda id, base_url: SimpleNamespace(id=id, base_url=base_url),
|
||||
)
|
||||
return provider, aio_mod
|
||||
|
||||
|
||||
def test_remote_warm_pool_revalidates_current_contract_before_reclaim(tmp_path, monkeypatch):
|
||||
provider, aio_mod = _make_remote_warm_provider(tmp_path, monkeypatch, "sandbox-warm")
|
||||
calls: list[dict] = []
|
||||
_install_remote_revalidation_backend(provider, aio_mod, calls)
|
||||
|
||||
result = provider._acquire_internal("thread-1", user_id="alice")
|
||||
|
||||
assert result == "sandbox-warm"
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["sandbox_id"] == "sandbox-warm"
|
||||
assert calls[0]["user_id"] == "alice"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_remote_warm_pool_revalidates_current_contract_before_reclaim_async(tmp_path, monkeypatch):
|
||||
provider, aio_mod = _make_remote_warm_provider(tmp_path, monkeypatch, "sandbox-warm-async")
|
||||
calls: list[dict] = []
|
||||
_install_remote_revalidation_backend(provider, aio_mod, calls)
|
||||
|
||||
result = await provider._acquire_internal_async("thread-1", user_id="alice")
|
||||
|
||||
assert result == "sandbox-warm-async"
|
||||
assert len(calls) == 1
|
||||
assert calls[0]["sandbox_id"] == "sandbox-warm-async"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_acquire_async_uses_async_readiness_polling(monkeypatch):
|
||||
"""AioSandboxProvider async creation must not use sync readiness polling."""
|
||||
@ -829,7 +1024,13 @@ def test_remote_backend_create_forwards_effective_user_id(monkeypatch):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return {"sandbox_url": "http://sandbox.local"}
|
||||
return {
|
||||
"sandbox_id": "sandbox-42",
|
||||
"sandbox_url": "http://sandbox.local",
|
||||
"user_id": "user-7",
|
||||
"thread_id": "thread-42",
|
||||
"mount_contract_version": 2,
|
||||
}
|
||||
|
||||
def _post(url, json, timeout, headers=None): # noqa: A002 - mirrors requests.post kwarg
|
||||
posted.update({"url": url, "json": json, "timeout": timeout})
|
||||
@ -866,7 +1067,13 @@ def test_remote_backend_create_prefers_explicit_user_id(monkeypatch):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return {"sandbox_url": "http://sandbox.local"}
|
||||
return {
|
||||
"sandbox_id": "sandbox-42",
|
||||
"sandbox_url": "http://sandbox.local",
|
||||
"user_id": "ou-user",
|
||||
"thread_id": "thread-42",
|
||||
"mount_contract_version": 2,
|
||||
}
|
||||
|
||||
def _post(url, json, timeout, headers=None): # noqa: A002 - mirrors requests.post kwarg
|
||||
posted.update({"url": url, "json": json, "timeout": timeout})
|
||||
|
||||
@ -204,6 +204,77 @@ def test_create_rejects_nondefault_user_on_unverified_legacy_provisioner(monkeyp
|
||||
assert called is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"response_payload",
|
||||
[
|
||||
{"sandbox_id": "abc123", "sandbox_url": "http://k3s:31001"},
|
||||
{
|
||||
"sandbox_id": "abc123",
|
||||
"sandbox_url": "http://k3s:31001",
|
||||
"user_id": "mallory",
|
||||
"thread_id": "thread-1",
|
||||
"mount_contract_version": 2,
|
||||
},
|
||||
{
|
||||
"sandbox_id": "abc123",
|
||||
"sandbox_url": "http://k3s:31001",
|
||||
"user_id": "alice",
|
||||
"thread_id": "other-thread",
|
||||
"mount_contract_version": 2,
|
||||
},
|
||||
{
|
||||
"sandbox_id": "abc123",
|
||||
"sandbox_url": "http://k3s:31001",
|
||||
"user_id": "alice",
|
||||
"thread_id": "thread-1",
|
||||
"mount_contract_version": 1,
|
||||
},
|
||||
],
|
||||
)
|
||||
def test_current_contract_create_rejects_unverified_or_mismatched_response(monkeypatch, response_payload):
|
||||
backend = RemoteSandboxBackend("http://provisioner:8002")
|
||||
backend._mount_contract_version = 2
|
||||
backend._mount_contract_capability_known = True
|
||||
monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
|
||||
monkeypatch.setattr(
|
||||
requests,
|
||||
"post",
|
||||
lambda *_args, **_kwargs: _StubResponse(payload=response_payload),
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="mount contract response"):
|
||||
backend.create("thread-1", "abc123", user_id="alice")
|
||||
|
||||
|
||||
def test_current_contract_create_returns_verified_identity(monkeypatch):
|
||||
backend = RemoteSandboxBackend("http://provisioner:8002")
|
||||
backend._mount_contract_version = 2
|
||||
backend._mount_contract_capability_known = True
|
||||
monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda _user_id: False)
|
||||
monkeypatch.setattr(
|
||||
requests,
|
||||
"post",
|
||||
lambda *_args, **_kwargs: _StubResponse(
|
||||
payload={
|
||||
"sandbox_id": "abc123",
|
||||
"sandbox_url": "http://k3s:31001",
|
||||
"user_id": "alice",
|
||||
"thread_id": "thread-1",
|
||||
"mount_contract_version": 2,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
info = backend.create("thread-1", "abc123", user_id="alice")
|
||||
|
||||
assert (info.sandbox_id, info.user_id, info.thread_id, info.mount_contract_version) == (
|
||||
"abc123",
|
||||
"alice",
|
||||
"thread-1",
|
||||
2,
|
||||
)
|
||||
|
||||
|
||||
def test_provisioner_create_returns_sandbox_info(monkeypatch):
|
||||
backend = RemoteSandboxBackend("http://provisioner:8002")
|
||||
monkeypatch.setattr(remote_backend_mod, "user_should_see_legacy_skills", lambda user_id: True)
|
||||
|
||||
@ -850,6 +850,71 @@ class TestCleanupStaleUploadStagingFiles:
|
||||
assert cleanup_stale_upload_staging_files(tmp_path) == 0
|
||||
assert staged.read_text() == "outside"
|
||||
|
||||
def test_restores_primary_left_in_deletion_transaction_after_crash(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"original")
|
||||
identity = UploadIdentity.from_path(primary)
|
||||
|
||||
staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
|
||||
uploads,
|
||||
primary,
|
||||
identity,
|
||||
)
|
||||
stage_lease.release()
|
||||
|
||||
assert not primary.exists()
|
||||
assert staged_path.read_bytes() == b"original"
|
||||
assert cleanup_stale_upload_staging_files(tmp_path) == 1
|
||||
assert primary.read_bytes() == b"original"
|
||||
assert not staged_path.exists()
|
||||
|
||||
def test_finishes_recovery_that_crashed_after_publishing_visible_link(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"original")
|
||||
identity = UploadIdentity.from_path(primary)
|
||||
staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
|
||||
uploads,
|
||||
primary,
|
||||
identity,
|
||||
)
|
||||
os.link(staged_path, primary)
|
||||
stage_lease.release()
|
||||
|
||||
assert cleanup_stale_upload_staging_files(tmp_path) == 1
|
||||
assert primary.read_bytes() == b"original"
|
||||
assert not staged_path.exists()
|
||||
|
||||
def test_refuses_replaced_deletion_tombstone_after_crash(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"original")
|
||||
identity = UploadIdentity.from_path(primary)
|
||||
staged_path, stage_lease = upload_manager_module._stage_primary_deletion(
|
||||
uploads,
|
||||
primary,
|
||||
identity,
|
||||
)
|
||||
replacement = tmp_path / "replacement.bin"
|
||||
replacement.write_bytes(b"replacement")
|
||||
staged_path.unlink()
|
||||
replacement.rename(staged_path)
|
||||
stage_lease.release()
|
||||
|
||||
assert cleanup_stale_upload_staging_files(tmp_path) == 0
|
||||
assert not primary.exists()
|
||||
assert staged_path.read_bytes() == b"replacement"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# delete_file_safe
|
||||
@ -1041,3 +1106,61 @@ class TestDeleteFileSafe:
|
||||
|
||||
assert primary.read_bytes() == b"PDF"
|
||||
assert owned_conversion.read_text(encoding="utf-8") == "generated"
|
||||
|
||||
def test_failed_delete_preserves_old_generation_when_name_is_recreated(self, tmp_path):
|
||||
uploads = tmp_path / "user-data" / "uploads"
|
||||
uploads.mkdir(parents=True)
|
||||
primary = uploads / "report.pdf"
|
||||
primary.write_bytes(b"old generation")
|
||||
|
||||
def recreate_then_fail(_filename):
|
||||
primary.write_bytes(b"new generation")
|
||||
raise OSError("remote delete failed")
|
||||
|
||||
with pytest.raises((OSError, UnsafeUploadPathError)):
|
||||
delete_file_safe(
|
||||
uploads,
|
||||
primary.name,
|
||||
delete_remote_copy=recreate_then_fail,
|
||||
)
|
||||
|
||||
recovered = list(uploads.glob("report.pdf_recovered_*"))
|
||||
assert primary.read_bytes() == b"new generation"
|
||||
assert len(recovered) == 1
|
||||
assert recovered[0].read_bytes() == b"old generation"
|
||||
assert cleanup_stale_upload_staging_files(tmp_path) == 0
|
||||
assert recovered[0].read_bytes() == b"old generation"
|
||||
|
||||
def test_portable_alias_rename_deletes_original_companions(self, tmp_path):
|
||||
import deerflow.uploads.manager as upload_manager_module
|
||||
|
||||
uploads = tmp_path / "user-data" / "uploads"
|
||||
uploads.mkdir(parents=True)
|
||||
primary = uploads / "straße.pdf"
|
||||
primary.write_bytes(b"payload")
|
||||
conversion = conversion_path_for_upload(primary)
|
||||
conversion.parent.mkdir()
|
||||
conversion.write_text("generated", encoding="utf-8")
|
||||
renamed = uploads / "strasse.pdf"
|
||||
remote_names: list[str] = []
|
||||
real_find = upload_manager_module._find_upload_path_by_identity
|
||||
|
||||
def rename_after_scan(base_dir, identity):
|
||||
found = real_find(base_dir, identity)
|
||||
found.rename(renamed)
|
||||
return renamed
|
||||
|
||||
with patch.object(
|
||||
upload_manager_module,
|
||||
"_find_upload_path_by_identity",
|
||||
side_effect=rename_after_scan,
|
||||
):
|
||||
delete_file_safe(
|
||||
uploads,
|
||||
primary.name,
|
||||
delete_remote_copy=remote_names.append,
|
||||
)
|
||||
|
||||
assert not renamed.exists()
|
||||
assert not conversion.exists()
|
||||
assert remote_names == ["straße.pdf"]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user