Upload 10 files

app.py

@@ -73,6 +73,7 @@ from router_comments import (router as comments_router, start_comment_writes, st
 from router_messages import (router as messages_router, start_message_io, stop_message_io, close_message_io) # ✉️ 私信系统
 from router_wallet import (router as wallet_router, start_wallet_sql, stop_wallet_sql, close_wallet_sql)     # 💰 钱包/提现
 from proxy_access import start_proxy_access, stop_proxy_access, close_proxy_access
+from proxy_download_work import start_proxy_downloads, stop_proxy_downloads, close_proxy_downloads
 from router_proxy import router as proxy_router       # 🔗 代理下载
 from router_posts import (router as posts_router, start_posts_io, stop_posts_io, close_posts_io)       # 💬 讨论区
 from router_prompts import (router as prompts_router, start_prompts_io, stop_prompts_io, close_prompts_io)   # ✨ 提示词市场
@@ -325,6 +326,7 @@ async def on_startup():
             start_prompts_io()
             start_tasks_io()
             start_proxy_access()
+            start_proxy_downloads()
             start_comment_reads()
             start_admin_scripts()
 
@@ -384,6 +386,7 @@ async def _shutdown_resources():
     failures.call('stop_warmup', stop_startup_warmup, app.state)
     for stage, stop in (
         ('stop_wallet_sql', stop_wallet_sql),
+        ('stop_proxy_downloads', stop_proxy_downloads),
         ('stop_resource_reads', stop_resource_reads), ('stop_comment_writes', stop_comment_writes),
         ('stop_posts', stop_posts_io), ('stop_prompts', stop_prompts_io), ('stop_tasks', stop_tasks_io), ('stop_proxy_access', stop_proxy_access),
         ('stop_messages', stop_message_io), ('stop_resources', stop_resource_work), ('stop_open_sql', stop_open_sql), ('stop_config_sql', stop_config_sql), ('stop_shared', stop_shared_work),
@@ -400,6 +403,7 @@ async def _shutdown_resources():
         failures.call('stop_' + name, request_periodic_stop, app.state, (name,))
     for stage, close, args in (
         ('wait_warmup', wait_startup_warmup, (app.state,)),
+        ('close_proxy_downloads', close_proxy_downloads, ()),
         ('close_wallet_sql', close_wallet_sql, ()),
         ('close_resource_reads', close_resource_reads, ()), ('close_comment_writes', close_comment_writes, ()),
         ('close_posts', close_posts_io, ()), ('close_prompts', close_prompts_io, ()), ('close_tasks', close_tasks_io, ()), ('close_proxy_access', close_proxy_access, ()),

github_zip_service.py

@@ -7,6 +7,7 @@ from skill_archive_source import source_header, installation_revision, source_ke
 from proxy_resources import ProxyResources
 from workflow_stream_response import OwnedProxyResponse
 from zip_stream import zip_response_length, stream_zip_bytes
+from owned_thread import run_owned_thread
 
 
 async def prepare_github_zip(item, item_id, *, store, token_selector, resolve_commit,
@@ -17,7 +18,7 @@ async def prepare_github_zip(item, item_id, *, store, token_selector, resolve_co
         source = github_item_source(item)
         repo_url = item.get("link", "")
         owner, repo = source["owner"], source["repo"]
-        active_token = token_selector(item)
+        active_token = await run_owned_thread(token_selector, item)
         skill_revision = installation_revision(item, source_key(source), active_token, signing_key=signing_key) if source['skill'] else None
     except ValueError:
         return error_response(content={"error": "资源仓库或凭据暂时无法核实"}, status_code=503)
@@ -26,7 +27,7 @@ async def prepare_github_zip(item, item_id, *, store, token_selector, resolve_co
     cache_resources = None
     cache_handed_off = False
     try:
-        versions_db = store.load_data("versions.json", default_data={})
+        versions_db = await run_owned_thread(store.load_data, "versions.json", default_data={})
         item_version = versions_db.get(item_id)
         archive = cached_archive(item, active_token, item_version, repo_id=store.DATASET_REPO_ID)
         hf_cache_token = os.environ.get("HF_TOKEN")

proxy_download_work.py

@@ -0,0 +1,189 @@
+"""下载准备、响应交付与应用关闭共用的有界归属。"""
+import asyncio
+from functools import wraps
+
+from fastapi import HTTPException
+from workflow_stream_response import OwnedProxyResponse
+
+
+def _spawn(function):
+    async def execute():
+        try:
+            return True, await function()
+        except BaseException as error:
+            return False, error
+    coroutine = execute()
+    try:
+        return asyncio.create_task(coroutine)
+    except BaseException:
+        coroutine.close()
+        raise
+
+
+async def _settle(task, interrupted=None):
+    cancelled = None
+    while not task.done():
+        try:
+            await asyncio.shield(task)
+        except asyncio.CancelledError as error:
+            cancelled = error
+            if interrupted is not None:
+                interrupted(task)
+    outcome = (False, asyncio.CancelledError()) if task.cancelled() else task.result()
+    return outcome, cancelled
+
+
+class _Download:
+    def __init__(self, group):
+        self.group = group
+        self.active = None
+        self.response = None
+        self.aborting = False
+        self.released = False
+        self.sending = False
+        self.cancelled_tasks = set()
+        self.cleanup = None
+        self.finished = asyncio.Event()
+
+    def cancel(self, task):
+        self.aborting = True
+        if not task.done() and task not in self.cancelled_tasks:
+            self.cancelled_tasks.add(task)
+            task.cancel()
+
+    def release(self):
+        self.released = True
+        self.group._pending.discard(self)
+        self.finished.set()
+
+    async def prepare(self, function, args, kwargs):
+        result = await function(*args, **kwargs)
+        if isinstance(result, OwnedProxyResponse):
+            result.bind_download(self)
+            self.response = result
+        return result
+
+    async def abort(self):
+        self.aborting = True
+        self.cancel(self.active)
+        await _settle(self.active, self.cancel)
+        if self.response is None:
+            self.release()
+            return
+        if self.sending:
+            # 发送拥有者先完成迭代器和连接关闭，再归还名额。
+            await _settle(_spawn(self.finished.wait))
+            return
+        if self.cleanup is None:
+            async def close():
+                try:
+                    await self.response.aclose()
+                finally:
+                    self.release()
+            self.cleanup = _spawn(close)
+        outcome, cancelled = await _settle(self.cleanup)
+        succeeded, result = outcome
+        if cancelled is not None:
+            raise cancelled
+        if not succeeded:
+            raise result
+
+    async def deliver(self, response, scope, receive, send):
+        if self.released or self.aborting or self.sending or response is not self.response:
+            raise asyncio.CancelledError()
+        self.sending = True
+        try:
+            self.active = _spawn(lambda: response.send_owned(scope, receive, send))
+            outcome, cancelled = await _settle(self.active, self.cancel)
+            succeeded, result = outcome
+            if cancelled is not None:
+                raise cancelled
+            if not succeeded:
+                raise result
+            return result
+        finally:
+            try:
+                # 发送任务在首次执行前被取消时，也必须关闭尚未消费的响应。
+                await response.aclose()
+            finally:
+                self.release()
+
+
+class ProxyDownloadWork:
+    def __init__(self, capacity=8):
+        if type(capacity) is not int or capacity < 1:
+            raise ValueError('invalid download capacity')
+        self.capacity = capacity
+        self._pending = set()
+        self._accepting = True
+        self._closing = None
+        self._closers = 0
+
+    def start(self):
+        if self._pending or self._closers or self._closing is not None and not self._closing.done():
+            raise RuntimeError('download work is still pending')
+        if self._closing is not None:
+            succeeded, result = self._closing.result()
+            if not succeeded:
+                raise result
+        self._closing = None
+        self._accepting = True
+
+    def stop(self):
+        self._accepting = False
+
+    def endpoint(self, function):
+        @wraps(function)
+        async def invoke(*args, **kwargs):
+            return await self.run(function, *args, **kwargs)
+        return invoke
+
+    async def run(self, function, *args, **kwargs):
+        if not self._accepting or len(self._pending) >= self.capacity:
+            raise HTTPException(status_code=503, detail='service_unavailable')
+        owner = _Download(self)
+        self._pending.add(owner)
+        handed_off = False
+        try:
+            owner.active = _spawn(lambda: owner.prepare(function, args, kwargs))
+            outcome, cancelled = await _settle(owner.active, owner.cancel)
+            succeeded, result = outcome
+            if cancelled is not None or owner.aborting:
+                raise cancelled or asyncio.CancelledError()
+            if not succeeded:
+                raise result
+            handed_off = owner.response is not None
+            return result
+        finally:
+            if owner.response is None:
+                owner.release()
+            elif not handed_off:
+                await owner.abort()
+
+    async def close(self):
+        self.stop()
+        self._closers += 1
+        try:
+            if self._closing is None:
+                async def drain():
+                    tasks = [_spawn(owner.abort) for owner in tuple(self._pending)]
+                    results = await asyncio.gather(*tasks)
+                    for succeeded, result in results:
+                        if not succeeded:
+                            raise result
+                self._closing = _spawn(drain)
+            outcome, cancelled = await _settle(self._closing)
+            succeeded, result = outcome
+            if cancelled is not None:
+                raise cancelled
+            if not succeeded:
+                raise result
+        finally:
+            self._closers -= 1
+
+
+_downloads = ProxyDownloadWork()
+proxy_download_endpoint = _downloads.endpoint
+start_proxy_downloads = _downloads.start
+stop_proxy_downloads = _downloads.stop
+close_proxy_downloads = _downloads.close

router_proxy.py

@@ -25,6 +25,8 @@ from zip_stream import ZIP_DOWNLOAD_MAX_BYTES
 from github_zip_service import prepare_github_zip
 from proxy_resources import close_redirect_response
 from skill_archive_source import install_source, SkillSourceError
+from proxy_download_work import proxy_download_endpoint
+from owned_thread import run_owned_thread
 
 router = APIRouter()
 MAX_PROXY_DOWNLOAD_BYTES = int(os.environ.get("MAX_PROXY_DOWNLOAD_BYTES", str(256 * 1024 * 1024)))
@@ -113,11 +115,12 @@ class SkillInstallSourceRequest(BaseModel):
 
 @router.post('/api/skill_install_source')
 @shared_limit(30)
+@proxy_download_endpoint
 async def skill_install_source(request: Request, req_data: SkillInstallSourceRequest,
                                account: str = Depends(require_auth)):
     try:
         item = await authorized_proxy_item(json_db, SessionLocal, req_data.item_id, account)
-        result = install_source(item, account, token_selector=select_github_token, signing_key=JWT_SECRET)
+        result = await run_owned_thread(install_source, item, account, token_selector=select_github_token, signing_key=JWT_SECRET)
     except ProxyAccessError as error:
         return JSONResponse(content={'error': error.message}, status_code=error.status_code,
                             headers={'Cache-Control': 'no-store'})
@@ -128,6 +131,7 @@ async def skill_install_source(request: Request, req_data: SkillInstallSourceReq
 
 @router.post("/api/proxy_github_zip")
 @shared_limit(5)
+@proxy_download_endpoint
 async def proxy_github_zip(request: Request, req_data: ProxyGithubZipRequest, account: str = Depends(require_auth)):
     """云端代理：校验所有权后，拉取 GitHub 私有库 ZIP 流透传给本地"""
     try:
@@ -190,6 +194,7 @@ class ProxyDownloadRequest(BaseModel):
 
 @router.post("/api/proxy_download")
 @shared_limit(10)
+@proxy_download_endpoint
 async def proxy_download(request: Request, req_data: ProxyDownloadRequest, account: str = Depends(require_auth)):
     """云端代理：校验所有权后，拉取真实的工作流 JSON 文件并透传给本地"""
     try:

workflow_download_service.py

@@ -5,6 +5,7 @@ from urllib.parse import urlparse, urljoin
 from workflow_decode import workflow_encoding, stream_workflow_bytes, workflow_response_length, WorkflowSizeError
 from workflow_stream_response import OwnedWorkflowResponse
 from proxy_resources import ProxyResources
+from owned_thread import run_owned_thread
 
 
 async def prepare_workflow_download(item, item_id, *, store, source_authorizer,
@@ -17,7 +18,7 @@ async def prepare_workflow_download(item, item_id, *, store, source_authorizer,
         return error_response(content={"error": "该资源没有可用下载地址"}, status_code=400)
 
     try:
-        workflow_path = await asyncio.to_thread(source_authorizer, store, item, target_url)
+        workflow_path = await run_owned_thread(source_authorizer, store, item, target_url)
     except source_error as error:
         return error_response(content={"error": error.message}, status_code=error.status_code)
     original_target = target_url

workflow_stream_response.py

@@ -1,4 +1,5 @@
 """代理流式响应的交付收尾；即使发送头/正文失败也归还所属连接。"""
+import asyncio
 from fastapi.responses import StreamingResponse
 
 
@@ -6,17 +7,44 @@ class OwnedProxyResponse(StreamingResponse):
     def __init__(self, content, *, close_callback, **kwargs):
         super().__init__(content, **kwargs)
         self._close_callback = close_callback
+        self._download = None
+        self._close_task = None
+
+    def bind_download(self, download):
+        if self._download is not None or self._close_task is not None:
+            raise RuntimeError('response is already owned or closed')
+        self._download = download
+
+    async def aclose(self):
+        if self._close_task is None:
+            async def close():
+                try:
+                    close_iterator = getattr(self.body_iterator, "aclose", None)
+                    if close_iterator is not None:
+                        await close_iterator()
+                finally:
+                    await self._close_callback()
+            self._close_task = asyncio.create_task(close())
+        cancelled = None
+        while not self._close_task.done():
+            try:
+                await asyncio.shield(self._close_task)
+            except asyncio.CancelledError as error:
+                cancelled = error
+        self._close_task.result()
+        if cancelled is not None:
+            raise cancelled
 
     async def __call__(self, scope, receive, send):
+        if self._download is not None:
+            return await self._download.deliver(self, scope, receive, send)
+        return await self.send_owned(scope, receive, send)
+
+    async def send_owned(self, scope, receive, send):
         try:
             await super().__call__(scope, receive, send)
         finally:
-            try:
-                close_iterator = getattr(self.body_iterator, "aclose", None)
-                if close_iterator is not None:
-                    await close_iterator()
-            finally:
-                await self._close_callback()
+            await self.aclose()
 
 # 保留既有工作流调用方的导出名称。
 OwnedWorkflowResponse = OwnedProxyResponse

云端_定时版本检测引擎.py

@@ -6,6 +6,7 @@ from zip_decode import stream_zip_archive_bytes
 from owned_thread import run_owned_thread
 from zip_validation import validate_zip_archive
 from proxy_resources import ProxyResources
+from skill_archive_source import read_metadata_response, METADATA_SECONDS
 # 云端_定时版本检测引擎.py
 import asyncio
 import datetime
@@ -49,32 +50,36 @@ def _normalize_version_entry(versions_db: dict, item_id: str):
 
 
 async def fetch_github_commit(repo_url, token, ref=None):
-    """
-    请求 GitHub API 获取最新版的 Commit Hash
-    🛡️ 稳定性优化：超时保护 + 异常捕获 + 重试机制
-    """
+    """在共同时间和正文预算内核实完整commit，失败不发布版本。"""
+    resources = None
     try:
         owner, repo = repository_parts(repo_url)
         
         api_url = f"https://api.github.com/repos/{owner}/{repo}/commits/" + commit_ref(ref) if ref is not None else f"https://api.github.com/repos/{owner}/{repo}/commits?per_page=1"
-        headers = {"Accept": "application/vnd.github.v3+json", "User-Agent": "ComfyUI-Hub"}
+        headers = {"Accept": "application/vnd.github.v3+json", "User-Agent": "ComfyUI-Hub",
+                   "Accept-Encoding": "identity"}
         if token:
             headers["Authorization"] = f"Bearer {token}"
         
-        # 🛡️ 稳定性优化：超时 + 重试
-        async with httpx.AsyncClient(follow_redirects=False, trust_env=False, transport=ValidatedAsyncHTTPTransport(), timeout=httpx.Timeout(15.0, connect=5.0)) as client:
-            for attempt in range(3):  # 最多重试 3 次
+        client = httpx.AsyncClient(follow_redirects=False, trust_env=False,
+            transport=ValidatedAsyncHTTPTransport(), timeout=httpx.Timeout(15.0, connect=5.0))
+        resources = ProxyResources(client)
+        async with asyncio.timeout(METADATA_SECONDS):
+            for attempt in range(3):
                 try:
-                    resp = await client.get(api_url, headers=headers)
+                    request = client.build_request('GET', api_url, headers=headers)
+                    resp = resources.track(await client.send(request, stream=True, follow_redirects=False))
                     if resp.status_code == 200:
-                        data = resp.json()
-                        latest_commit = data[0] if ref is None and isinstance(data, list) and data else data
+                        # per_page=1只使用本页，不访问分页地址。
+                        data = await read_metadata_response(resp, resources, allow_pagination=True)
+                        latest_commit = data[0] if ref is None and isinstance(data, list) and len(data) == 1 else data if ref is not None else None
                         if isinstance(latest_commit, dict):
                             sha = commit_sha(latest_commit.get("sha"))
                             if ref is not None and not sha.startswith(commit_ref(ref)):
                                 return None
                             return sha
-                    elif resp.status_code == 403:  # Rate limit
+                    await resources.close_response(resp)
+                    if resp.status_code == 403:
                         logger.warning("GitHub API 速率限制: status=403")
                         return None
                     break
@@ -85,6 +90,9 @@ async def fetch_github_commit(repo_url, token, ref=None):
                     logger.warning(f"版本检测请求失败: error_type={type(e).__name__}")
     except Exception as e:
         logger.error(f"检测版本失败: error_type={type(e).__name__}")
+    finally:
+        if resources is not None:
+            await resources.close()
     return None
 
 async def fetch_latest_github_hash(repo_url, token):