diff --git a/app/api.py b/app/api.py index 0cc1411..2b4cf01 100644 --- a/app/api.py +++ b/app/api.py @@ -44,10 +44,13 @@ from app.schemas import ( NodeView, IPWhitelistCreate, IPWhitelistOut, + BanToggle, WhitelistToggle, RegisterIn, SettleOut, SummaryOut, + AdminTaskOut, + TaskCreate, TaskOut, TaskReportIn, TokenOut, @@ -69,7 +72,7 @@ from app.services.common import get_node_by_sn, get_user_by_wallet from app.services.datacenter import approve_application, list_applications, reject_application, submit_application from app.services.hardware import HardwareReport from app.services.hardware import display_node_id, displayed_population -from app.services.heartbeat import assert_node_token, record_heartbeat +from app.services.heartbeat import assert_node_token, record_heartbeat, set_node_banned from app.services.ip_guard import add_whitelist_ip, client_ip, list_whitelist, remove_whitelist_ip from app.services.orders import approve_order, list_orders, place_order, reject_order, render_client from app.services.portal import ( @@ -81,7 +84,7 @@ from app.services.portal import ( user_rows, ) from app.services.registry import register_node, register_user -from app.services.tasks import list_pending_tasks, report_task +from app.services.tasks import create_admin_tasks, list_admin_tasks, list_pending_tasks, report_task, result_preview from app.services.withdrawals import available_balance, confirm_withdrawal, reject_withdrawal, request_withdrawal from app.services.yield_engine import compute_node_yield, list_node_yields, network_yields, settle_yield_date @@ -210,6 +213,14 @@ def admin_toggle_node_whitelist(sn: str, body: WhitelistToggle, _: AdminUser, db return NodeView.model_validate(snapshot) +@router.post("/admin/nodes/{sn}/ban", response_model=NodeView) +def admin_ban_node(sn: str, body: BanToggle, _: AdminUser, db: Db) -> NodeView: + node = get_node_by_sn(db, sn) + set_node_banned(db, node, body.is_banned) + snapshot = next(item for item in live_nodes(db) if item["sn"] == node.sn) + return NodeView.model_validate(snapshot) + + @router.get("/admin/ip-whitelist", response_model=list[IPWhitelistOut]) def admin_ip_whitelist(_: AdminUser, db: Db) -> list[IPWhitelist]: return list_whitelist(db) @@ -341,7 +352,18 @@ def pending_tasks(sn: str, db: Db, token: str | None = None) -> list[TaskOut]: @router.post("/node/task/report", response_model=TaskOut) def report(body: TaskReportIn, db: Db) -> TaskOut: assert_node_token(get_node_by_sn(db, body.sn), body.token) - return _task_out(report_task(db, body.sn, body.task_id, body.success, body.nonce)) + return _task_out(report_task(db, body.sn, body.task_id, body.success, body.nonce, result=body.result)) + + +@router.post("/admin/tasks/create", response_model=list[AdminTaskOut]) +def admin_create_tasks(body: TaskCreate, _: AdminUser, db: Db) -> list[AdminTaskOut]: + tasks = create_admin_tasks(db, body.task_type, body.payload, body.target_node_sn) + return [_admin_task_out(task) for task in tasks] + + +@router.get("/admin/tasks/list", response_model=list[AdminTaskOut]) +def admin_list_tasks(_: AdminUser, db: Db) -> list[AdminTaskOut]: + return [_admin_task_out(task) for task in list_admin_tasks(db)] @router.get("/nodes/{sn}/contribution", response_model=ContributionOut) @@ -487,6 +509,23 @@ def _withdrawal_view(row: WithdrawalLog) -> WithdrawalView: ) +def _admin_task_out(task: ComputeTask) -> AdminTaskOut: + node = task.node + return AdminTaskOut( + task_id=task.id, + task_type=task.task_type, + target_node_sn=node.sn, + display_node_id=node.display_node_id, + status=task.status, + payload=task.payload or {}, + result=task.result, + result_preview=result_preview(task.result), + dispatched_at=task.dispatched_at, + deadline_at=task.deadline_at, + responded_at=task.responded_at, + ) + + def _task_out(task: ComputeTask) -> TaskOut: return TaskOut( task_id=task.id, diff --git a/app/models.py b/app/models.py index 0d8258a..dede817 100644 --- a/app/models.py +++ b/app/models.py @@ -90,6 +90,7 @@ class WithdrawalStatus(str, enum.Enum): class TaskStatus(str, enum.Enum): PENDING = "pending" + PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" EXPIRED = "expired" @@ -169,6 +170,7 @@ class ComputeNode(Base): generation_locked: Mapped[bool] = mapped_column(default=False, comment="世代已按上报显卡锁定") recoverable_stake_amount: Mapped[Decimal | None] = mapped_column(MONEY, comment="尚未返还的质押金") is_whitelist: Mapped[bool] = mapped_column(Boolean, default=False, comment="白名单节点,跳过单 IP 限制") + is_banned: Mapped[bool] = mapped_column(Boolean, default=False, comment="封禁后拒绝心跳") client_ip: Mapped[str | None] = mapped_column(String(64), index=True, comment="最近一次注册或心跳的客户端 IP") hardware_fingerprint: Mapped[str | None] = mapped_column( String(64), unique=True, comment="显卡 UUID、主板与网卡合成的硬件指纹" @@ -257,6 +259,7 @@ class ComputeTask(Base): deadline_at: Mapped[datetime] = mapped_column(UtcDateTime) responded_at: Mapped[datetime | None] = mapped_column(UtcDateTime) success: Mapped[bool | None] = mapped_column() + result: Mapped[dict | None] = mapped_column(JSON, comment="节点回传的任务结果") node: Mapped[ComputeNode] = relationship(back_populates="tasks") diff --git a/app/schema_upgrade.py b/app/schema_upgrade.py index 70cdbb3..048f371 100644 --- a/app/schema_upgrade.py +++ b/app/schema_upgrade.py @@ -20,6 +20,10 @@ def upgrade_schema() -> None: if "role" not in columns: statements.append("ALTER TABLE users ADD COLUMN role VARCHAR(16) DEFAULT 'user' NOT NULL") need_username_index = not _has_username_index(inspector) + if "compute_tasks" in tables: + task_columns = {column["name"] for column in inspector.get_columns("compute_tasks")} + if "result" not in task_columns: + statements.append("ALTER TABLE compute_tasks ADD COLUMN result JSON") if "compute_nodes" in tables: node_columns = {column["name"] for column in inspector.get_columns("compute_nodes")} if "node_token" not in node_columns: @@ -53,6 +57,10 @@ def upgrade_schema() -> None: statements.append( "ALTER TABLE compute_nodes ADD COLUMN is_whitelist BOOLEAN DEFAULT 0 NOT NULL" ) + if "is_banned" not in node_columns: + statements.append( + "ALTER TABLE compute_nodes ADD COLUMN is_banned BOOLEAN DEFAULT 0 NOT NULL" + ) if "client_ip" not in node_columns: statements.append("ALTER TABLE compute_nodes ADD COLUMN client_ip VARCHAR(64)") statements.append( diff --git a/app/schemas.py b/app/schemas.py index 3b8048a..6b151b2 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -93,6 +93,7 @@ class NodeOut(APIModel): generation_locked: bool = False recoverable_stake_amount: MoneyText | None = None is_whitelist: bool = False + is_banned: bool = False client_ip: str | None = None hardware_fingerprint: str | None = None @@ -144,6 +145,27 @@ class TaskReportIn(APIModel): success: bool nonce: str = Field(min_length=8, max_length=128) token: str | None = Field(default=None, max_length=128) + result: dict | None = None + + +class TaskCreate(APIModel): + task_type: str = Field(min_length=1, max_length=32) + target_node_sn: str | None = Field(default=None, max_length=64) + payload: dict = Field(default_factory=dict) + + +class AdminTaskOut(APIModel): + task_id: str + task_type: str + target_node_sn: str + display_node_id: int + status: TaskStatus + payload: dict + result: dict | None = None + result_preview: str = "" + dispatched_at: datetime + deadline_at: datetime + responded_at: datetime | None = None class ContributionOut(APIModel): @@ -293,6 +315,10 @@ class WhitelistToggle(APIModel): is_whitelist: bool +class BanToggle(APIModel): + is_banned: bool + + class IPWhitelistCreate(APIModel): ip_address: str = Field(min_length=3, max_length=64) note: str | None = Field(default=None, max_length=128) diff --git a/app/services/heartbeat.py b/app/services/heartbeat.py index 8ae1626..0167a3a 100644 --- a/app/services/heartbeat.py +++ b/app/services/heartbeat.py @@ -40,6 +40,8 @@ def record_heartbeat( moment = now or utcnow() node = get_node_by_sn(session, sn.strip()) assert_node_token(node, token) + if node.is_banned: + raise AppError("节点已封禁", 403) if node.license_expires_at <= moment: raise AppError("节点授权已到期", 403) bound_ip = (client_ip or "").strip() or None @@ -65,6 +67,18 @@ def record_heartbeat( return node +def set_node_banned(session: Session, node: ComputeNode, banned: bool, now: datetime | None = None) -> ComputeNode: + """封禁立即结束在线区间。解除后要等下一次心跳才会重新上线。""" + moment = now or utcnow() + node.is_banned = banned + node.updated_at = moment + if banned and node.status == NodeStatus.ONLINE: + _close_open_intervals(session, node.id, moment, "banned") + node.status = NodeStatus.OFFLINE + session.flush() + return node + + def mark_stale_nodes(session: Session, now: datetime | None = None) -> list[str]: """把超过心跳时限或授权已到期的在线节点置为 offline,并关闭在线区间。""" moment = now or utcnow() diff --git a/app/services/portal.py b/app/services/portal.py index f64836d..8034f1c 100644 --- a/app/services/portal.py +++ b/app/services/portal.py @@ -82,6 +82,7 @@ def live_nodes(session: Session, user_id: int | None = None) -> list[dict]: "generation_locked": bool(node.generation_locked), "recoverable_stake_amount": node.recoverable_stake_amount, "is_whitelist": bool(node.is_whitelist), + "is_banned": bool(node.is_banned), "client_ip": node.client_ip, "hardware_fingerprint": node.hardware_fingerprint, } diff --git a/app/services/tasks.py b/app/services/tasks.py index 87f55dc..5751c07 100644 --- a/app/services/tasks.py +++ b/app/services/tasks.py @@ -18,7 +18,7 @@ from urllib.error import URLError from urllib.request import Request, urlopen from sqlalchemy import select -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, selectinload from app.clock import utcnow from app.config import settings @@ -28,13 +28,105 @@ from app.services.common import get_node_by_sn logger = logging.getLogger(__name__) +ADMIN_TASK_TYPES = { + "alpha_mining": "量化因子挖掘", + "llm_inference": "大模型推理", + "lora_training": "QLoRA 微调", + "stress_test": "简易压测", +} +_BUSY = (TaskStatus.PENDING, TaskStatus.PROCESSING) + + +def create_admin_tasks( + session: Session, + task_type: str, + payload: dict | None, + target_node_sn: str | None = None, + now: datetime | None = None, +) -> list[ComputeTask]: + """向指定节点或当前全部在线节点各放一条任务。payload 里补上一次性 nonce。""" + if task_type not in ADMIN_TASK_TYPES: + raise AppError("不支持的任务类型", 400) + moment = now or utcnow() + body = dict(payload or {}) + if len(json.dumps(body, ensure_ascii=False)) > 8000: + raise AppError("任务参数过长", 400) + targets = _dispatch_targets(session, target_node_sn, moment) + deadline = moment + timedelta(minutes=settings.task_deadline_minutes) + tasks: list[ComputeTask] = [] + for node in targets: + task_payload = dict(body) + task_payload["nonce"] = secrets.token_hex(16) + task = ComputeTask( + id=str(uuid.uuid4()), + node_id=node.id, + task_type=task_type, + payload=task_payload, + status=TaskStatus.PENDING, + push_status=PushStatus.QUEUED, + dispatched_at=moment, + deadline_at=deadline, + ) + session.add(task) + tasks.append(task) + session.flush() + for node, task in zip(targets, tasks): + _push_task(node, task) + session.flush() + return tasks + + +def list_admin_tasks(session: Session) -> list[ComputeTask]: + return list( + session.scalars( + select(ComputeTask) + .options(selectinload(ComputeTask.node)) + .order_by(ComputeTask.dispatched_at.desc(), ComputeTask.id.desc()) + ).all() + ) + + +def result_preview(result: dict | None) -> str: + if not result: + return "" + for key in ("formula", "alpha", "output", "summary"): + value = result.get(key) + if isinstance(value, str) and value.strip(): + return value.strip()[:120] + for key in ("patch_bytes", "patch_size", "adapter_bytes"): + if result.get(key) is not None: + return f"补丁 {result[key]}" + text = json.dumps(result, ensure_ascii=False) + return text if len(text) <= 120 else text[:117] + "..." + + +def _dispatch_targets(session: Session, target_node_sn: str | None, moment: datetime) -> list[ComputeNode]: + sn = (target_node_sn or "").strip() + if sn: + node = get_node_by_sn(session, sn) + if node.is_banned: + raise AppError("节点已封禁,不能下发任务", 409) + if node.license_expires_at <= moment: + raise AppError("节点授权已到期,不能下发任务", 409) + return [node] + nodes = session.scalars( + select(ComputeNode).where( + ComputeNode.status == NodeStatus.ONLINE, + ComputeNode.license_expires_at > moment, + ComputeNode.is_banned.is_(False), + ) + ).all() + if not nodes: + raise AppError("当前没有可下发的在线节点", 409) + return list(nodes) + def dispatch_random_tasks(session: Session, now: datetime | None = None) -> list[ComputeTask]: """向当前在线、授权仍有效、且没有未完成任务的节点随机下发一轮任务。""" moment = now or utcnow() expire_overdue_tasks(session, moment) pending_node_ids = set( - session.scalars(select(ComputeTask.node_id).where(ComputeTask.status == TaskStatus.PENDING)).all() + session.scalars(select(ComputeTask.node_id).where(ComputeTask.status.in_(_BUSY))).all() ) candidates = [ node @@ -90,6 +182,7 @@ def list_pending_tasks(session: Session, sn: str, now: datetime | None = None) - if task.deadline_at < moment: task.status = TaskStatus.EXPIRED continue + task.status = TaskStatus.PROCESSING visible.append(task) session.flush() return visible @@ -102,6 +195,7 @@ def report_task( success: bool, nonce: str, now: datetime | None = None, + result: dict | None = None, ) -> ComputeTask: """校验任务归属、时限和 nonce。只有校验通过且 success 为真才记为有效响应。""" moment = now or utcnow() @@ -111,7 +205,7 @@ def report_task( raise AppError("任务不存在", 404) if task.node_id != node.id: raise AppError("任务不属于该节点", 403) - if task.status != TaskStatus.PENDING: + if task.status not in _BUSY: raise AppError("任务已结束,不能重复回传", 409) if task.deadline_at < moment: task.status = TaskStatus.EXPIRED @@ -124,6 +218,7 @@ def report_task( task.responded_at = moment task.success = success + task.result = result or None task.status = TaskStatus.COMPLETED if success else TaskStatus.FAILED session.flush() return task @@ -133,7 +228,7 @@ def expire_overdue_tasks(session: Session, now: datetime | None = None) -> int: moment = now or utcnow() tasks = session.scalars( select(ComputeTask).where( - ComputeTask.status == TaskStatus.PENDING, + ComputeTask.status.in_(_BUSY), ComputeTask.deadline_at < moment, ) ).all() diff --git a/frontend/src/components/Shell.vue b/frontend/src/components/Shell.vue index 7b1a18b..4fc9c29 100644 --- a/frontend/src/components/Shell.vue +++ b/frontend/src/components/Shell.vue @@ -1,6 +1,6 @@ - + +