From 2bbdb00a4d835afa753acf49c2291697891ba486 Mon Sep 17 00:00:00 2001 From: dekun Date: Sat, 10 Oct 2026 21:53:20 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=20HyperNode-AI=20?= =?UTF-8?q?=E6=8E=A7=E5=88=B6=E9=9D=A2=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 包含节点账本、每日清算，以及 Ubuntu 交互式部署脚本。 Co-authored-by: Cursor --- .gitattributes | 2 + .gitignore | 7 + app/__init__.py | 1 + app/__main__.py | 6 + app/api.py | 132 ++++++++++++++++ app/clock.py | 27 ++++ app/config.py | 31 ++++ app/database.py | 71 +++++++++ app/errors.py | 8 + app/jobs/__init__.py | 1 + app/jobs/daily_yield.py | 41 +++++ app/main.py | 41 +++++ app/models.py | 251 ++++++++++++++++++++++++++++++ app/money.py | 32 ++++ app/scheduler.py | 107 +++++++++++++ app/schemas.py | 156 +++++++++++++++++++ app/services/__init__.py | 1 + app/services/common.py | 21 +++ app/services/heartbeat.py | 78 ++++++++++ app/services/locks.py | 41 +++++ app/services/registry.py | 74 +++++++++ app/services/tasks.py | 183 ++++++++++++++++++++++ app/services/withdrawals.py | 99 ++++++++++++ app/services/yield_engine.py | 236 ++++++++++++++++++++++++++++ app/settle.py | 35 +++++ deploy/ubuntu22.sh | 264 +++++++++++++++++++++++++++++++ requirements.txt | 8 + tests/__init__.py | 0 tests/test_core_flow.py | 294 +++++++++++++++++++++++++++++++++++ 说明.md | 110 +++++++++++++ 部署.md | 92 +++++++++++ 31 files changed, 2450 insertions(+) create mode 100644 .gitattributes create mode 100644 .gitignore create mode 100644 app/__init__.py create mode 100644 app/__main__.py create mode 100644 app/api.py create mode 100644 app/clock.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/errors.py create mode 100644 app/jobs/__init__.py create mode 100644 app/jobs/daily_yield.py create mode 100644 app/main.py create mode 100644 app/models.py create mode 100644 app/money.py create mode 100644 app/scheduler.py create mode 100644 app/schemas.py create mode 100644 app/services/__init__.py create mode 100644 app/services/common.py create mode 100644 app/services/heartbeat.py create mode 100644 app/services/locks.py create mode 100644 app/services/registry.py create mode 100644 app/services/tasks.py create mode 100644 app/services/withdrawals.py create mode 100644 app/services/yield_engine.py create mode 100644 app/settle.py create mode 100644 deploy/ubuntu22.sh create mode 100644 requirements.txt create mode 100644 tests/__init__.py create mode 100644 tests/test_core_flow.py create mode 100644 说明.md create mode 100644 部署.md diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..efdba87 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +* text=auto +*.sh text eol=lf diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4153ad9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +.venv/ +venv/ +.env +*.db +.pytest_cache/ diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..832d97b --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""DePIN 算力节点与任务调度控制面。""" diff --git a/app/__main__.py b/app/__main__.py new file mode 100644 index 0000000..8010b9d --- /dev/null +++ b/app/__main__.py @@ -0,0 +1,6 @@ +"""python -m app 启动 API。每日清算请使用 python -m app.settle。""" + +import uvicorn + +if __name__ == "__main__": + uvicorn.run("app.main:app", host="0.0.0.0", port=8000) diff --git a/app/api.py b/app/api.py new file mode 100644 index 0000000..ca63aa6 --- /dev/null +++ b/app/api.py @@ -0,0 +1,132 @@ +"""对外 HTTP 接口。业务规则在 services 中，这里只做参数转发。""" + +from datetime import date, timedelta +from typing import Annotated + +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.clock import utcnow +from app.database import get_db +from app.models import ComputeTask +from app.schemas import ( + BalanceOut, + ContributionOut, + HeartbeatIn, + HeartbeatOut, + NodeCreate, + NodeOut, + SettleOut, + TaskOut, + TaskReportIn, + UserCreate, + UserOut, + WithdrawalConfirm, + WithdrawalCreate, + WithdrawalOut, + WithdrawalReject, + YieldOut, +) +from app.services.common import get_node_by_sn, get_user_by_wallet +from app.services.heartbeat import record_heartbeat +from app.services.registry import register_node, register_user +from app.services.tasks import list_pending_tasks, report_task +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, settle_yield_date + +router = APIRouter(prefix="/api/v1") +Db = Annotated[Session, Depends(get_db)] + + +@router.post("/users", response_model=UserOut) +def create_user(body: UserCreate, db: Db) -> UserOut: + return register_user(db, body.wallet_address) + + +@router.post("/nodes", response_model=NodeOut) +def create_node(body: NodeCreate, db: Db) -> NodeOut: + return register_node(db, body.wallet_address, body.sn, body.purchase_amount, body.callback_url) + + +@router.get("/nodes/{sn}", response_model=NodeOut) +def get_node(sn: str, db: Db) -> NodeOut: + return get_node_by_sn(db, sn) + + +@router.post("/node/heartbeat", response_model=HeartbeatOut) +def heartbeat(body: HeartbeatIn, db: Db) -> HeartbeatOut: + node = record_heartbeat(db, body.sn) + return HeartbeatOut( + sn=node.sn, + status=node.status, + last_heartbeat_at=node.last_heartbeat_at, + server_time=node.last_heartbeat_at, + ) + + +@router.get("/node/{sn}/tasks", response_model=list[TaskOut]) +def pending_tasks(sn: str, db: Db) -> list[TaskOut]: + return [_task_out(task) for task in list_pending_tasks(db, sn)] + + +@router.post("/node/task/report", response_model=TaskOut) +def report(body: TaskReportIn, db: Db) -> TaskOut: + return _task_out(report_task(db, body.sn, body.task_id, body.success, body.nonce)) + + +@router.get("/nodes/{sn}/contribution", response_model=ContributionOut) +def contribution(sn: str, yield_date: date, db: Db) -> ContributionOut: + node = get_node_by_sn(db, sn) + computed = compute_node_yield(db, node, yield_date) + return ContributionOut(sn=node.sn, **computed.__dict__) + + +@router.post("/yield/settle", response_model=SettleOut) +def settle(db: Db, yield_date: date | None = None) -> SettleOut: + target = yield_date or (utcnow().date() - timedelta(days=1)) + summary = settle_yield_date(db, target) + return SettleOut( + yield_date=summary.yield_date, + created=summary.created, + skipped=summary.skipped, + total_points=summary.total_points, + is_partial=summary.is_partial, + ) + + +@router.get("/nodes/{sn}/yields", response_model=list[YieldOut]) +def yields(sn: str, db: Db) -> list[YieldOut]: + return list_node_yields(db, sn) + + +@router.get("/users/{wallet_address}/balance", response_model=BalanceOut) +def balance(wallet_address: str, db: Db) -> BalanceOut: + user = get_user_by_wallet(db, wallet_address) + return BalanceOut(wallet_address=user.wallet_address, available_points=available_balance(db, user.id)) + + +@router.post("/withdrawals", response_model=WithdrawalOut) +def create_withdrawal(body: WithdrawalCreate, db: Db) -> WithdrawalOut: + row, remaining = request_withdrawal(db, body.wallet_address, body.amount, body.asset_symbol) + return WithdrawalOut.model_validate(row).model_copy(update={"available_after": remaining}) + + +@router.post("/withdrawals/{withdrawal_id}/confirm", response_model=WithdrawalOut) +def confirm(withdrawal_id: int, body: WithdrawalConfirm, db: Db) -> WithdrawalOut: + return confirm_withdrawal(db, withdrawal_id, body.tx_hash) + + +@router.post("/withdrawals/{withdrawal_id}/reject", response_model=WithdrawalOut) +def reject(withdrawal_id: int, body: WithdrawalReject, db: Db) -> WithdrawalOut: + return reject_withdrawal(db, withdrawal_id, body.note) + + +def _task_out(task: ComputeTask) -> TaskOut: + return TaskOut( + task_id=task.id, + task_type=task.task_type, + payload=task.payload, + status=task.status, + dispatched_at=task.dispatched_at, + deadline_at=task.deadline_at, + ) diff --git a/app/clock.py b/app/clock.py new file mode 100644 index 0000000..64c787d --- /dev/null +++ b/app/clock.py @@ -0,0 +1,27 @@ +"""UTC 时间工具。账本、心跳和授权窗口全部使用 UTC。""" + +from __future__ import annotations + +import calendar +from datetime import date, datetime, time, timedelta, timezone + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def day_bounds(value: date) -> tuple[datetime, datetime]: + """返回某个 UTC 日期的左闭右开区间。""" + start = datetime.combine(value, time.min, tzinfo=timezone.utc) + return start, start + timedelta(days=1) + + +def add_months(value: datetime, months: int) -> datetime: + """按日历月平移，月末日期自动收敛到目标月最后一天。""" + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + month_index = value.month - 1 + months + year = value.year + month_index // 12 + month = month_index % 12 + 1 + day = min(value.day, calendar.monthrange(year, month)[1]) + return value.replace(year=year, month=month, day=day) diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..4bfd59a --- /dev/null +++ b/app/config.py @@ -0,0 +1,31 @@ +"""运行参数。可用环境变量或项目根目录的 .env 覆盖。""" + +from decimal import Decimal + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + database_url: str = "sqlite:///./depin.db" + # 超过该时长没有心跳，节点会被置为 offline。 + heartbeat_timeout_minutes: int = 15 + # 在线巡检周期。 + offline_check_minutes: int = 5 + # 随机任务下发周期、单轮最多下发数量，以及节点必须回传的时限。 + task_dispatch_minutes: int = 5 + dispatch_batch_size: int = 20 + task_deadline_minutes: int = 10 + # 当日既有在线样本又有任务样本时的权重。两者默认相加为 1。 + online_weight: Decimal = Decimal("0.7") + task_weight: Decimal = Decimal("0.3") + # 每日清算时刻。默认北京时间凌晨 00:05，入账的是上一个已结束的 UTC 账本日。 + yield_timezone: str = "Asia/Shanghai" + yield_hour: int = 0 + yield_minute: int = 5 + license_months: int = 12 + disable_scheduler: bool = False + + model_config = SettingsConfigDict(env_file=".env", extra="ignore") + + +settings = Settings() diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..6f88cd1 --- /dev/null +++ b/app/database.py @@ -0,0 +1,71 @@ +"""数据库引擎与会话。SQLite 仅用于本地启动，生产环境把 DATABASE_URL 指到 PostgreSQL。""" + +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager +from datetime import datetime, timezone + +from sqlalchemy import DateTime, create_engine +from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker +from sqlalchemy.types import TypeDecorator + +from app.config import settings + + +class UtcDateTime(TypeDecorator): + """统一读写 UTC，避免 SQLite 把时区信息丢掉后比较失败。""" + + impl = DateTime + cache_ok = True + + def process_bind_param(self, value, dialect): + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + def process_result_value(self, value, dialect): + if value is None: + return None + if isinstance(value, str): + text = value.replace("Z", "+00:00") + value = datetime.fromisoformat(text) + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +class Base(DeclarativeBase): + pass + + +connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {} +engine = create_engine(settings.database_url, connect_args=connect_args, pool_pre_ping=True) +SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False) + + +@contextmanager +def session_scope() -> Iterator[Session]: + session = SessionLocal() + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() + + +def get_db() -> Iterator[Session]: + session = SessionLocal() + try: + yield session + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() diff --git a/app/errors.py b/app/errors.py new file mode 100644 index 0000000..5f59dd4 --- /dev/null +++ b/app/errors.py @@ -0,0 +1,8 @@ +"""业务异常。路由层把它转换成对应的 HTTP 状态码。""" + + +class AppError(Exception): + def __init__(self, message: str, status_code: int = 400) -> None: + super().__init__(message) + self.message = message + self.status_code = status_code diff --git a/app/jobs/__init__.py b/app/jobs/__init__.py new file mode 100644 index 0000000..e32e1b3 --- /dev/null +++ b/app/jobs/__init__.py @@ -0,0 +1 @@ +"""后台定时任务。""" diff --git a/app/jobs/daily_yield.py b/app/jobs/daily_yield.py new file mode 100644 index 0000000..5a4ac16 --- /dev/null +++ b/app/jobs/daily_yield.py @@ -0,0 +1,41 @@ +"""每天凌晨的积分清算任务。""" + +from __future__ import annotations + +import logging +from datetime import date, datetime, timedelta, timezone + +from app.clock import utcnow +from app.database import session_scope +from app.services.heartbeat import mark_stale_nodes +from app.services.yield_engine import SettleSummary, settle_yield_date + +logger = logging.getLogger(__name__) + + +def run_daily_yield_cron(now: datetime | None = None, yield_date: date | None = None) -> SettleSummary: + """计算所有仍在授权期内的节点，并把当日积分写入 node_yields。 + + 调度器在北京时间 00:05 调用本函数。账本按 UTC 日切分，因此入账日期是 + 调用时刻之前最近一个已经结束的 UTC 日，而不是刚开始的这一天。 + 同一节点同一天已有流水时跳过，重复触发不会重复入账。 + """ + moment = now or utcnow() + if moment.tzinfo is None: + raise ValueError("清算时间必须带时区") + target = yield_date or (moment.astimezone(timezone.utc).date() - timedelta(days=1)) + + with session_scope() as session: + stale = mark_stale_nodes(session, moment) + summary = settle_yield_date(session, target, moment) + + logger.info( + "daily yield settled date=%s created=%s skipped=%s points=%s partial=%s stale_nodes=%s", + summary.yield_date, + summary.created, + summary.skipped, + format(summary.total_points, "f"), + summary.is_partial, + len(stale), + ) + return summary diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..b4255e3 --- /dev/null +++ b/app/main.py @@ -0,0 +1,41 @@ +"""应用入口。""" + +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +import app.models # noqa: F401 注册全部表 +from app.api import router +from app.database import Base, engine +from app.errors import AppError +from app.scheduler import start_scheduler, stop_scheduler + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + + +@asynccontextmanager +async def lifespan(_: FastAPI): + Base.metadata.create_all(bind=engine) + start_scheduler() + try: + yield + finally: + stop_scheduler() + + +app = FastAPI(title="DePIN 算力节点与任务调度", version="1.0.0", lifespan=lifespan) +app.include_router(router) + + +@app.exception_handler(AppError) +async def handle_app_error(_: Request, exc: AppError) -> JSONResponse: + return JSONResponse(status_code=exc.status_code, content={"detail": exc.message}) + + +@app.get("/health") +def health() -> dict[str, str]: + return {"status": "ok"} diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..dc0220a --- /dev/null +++ b/app/models.py @@ -0,0 +1,251 @@ +"""DePIN 控制面数据模型。 + +核心表： +- User：用户与钱包地址 +- ComputeNode：算力节点、授权金额、12 个月授权周期、在线状态 +- NodeYield：每日算力产出与积分流水 +- WithdrawalLog：数字资产提现申请、链上哈希与状态 + +支撑表（业务计算所依赖，不单独对外当作资产账本）： +- UptimeInterval：一次连续在线区间，用来计算有效在线时长 +- ComputeTask：下发给节点的模拟计算任务及回传校验结果 +- SchedulerLock：多实例部署时，定时任务靠数据库互斥，避免重复下发和重复清算 +""" + +from __future__ import annotations + +import enum +from datetime import date, datetime +from decimal import Decimal + +from sqlalchemy import ( + CheckConstraint, + Enum, + ForeignKey, + Index, + Integer, + Numeric, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy.types import JSON, TypeDecorator + +from app.clock import utcnow +from app.database import Base, UtcDateTime +from app.money import money + + +class PreciseDecimal(TypeDecorator): + """金额与比率。SQLite 会把 Numeric 读成 float，这里统一还原成 8 位小数。""" + + impl = Numeric(24, 8) + cache_ok = True + + def process_bind_param(self, value, dialect): + if value is None: + return None + return money(value) + + def process_result_value(self, value, dialect): + if value is None: + return None + return money(value) + + +MONEY = PreciseDecimal() +RATE = PreciseDecimal() + + +def _enum(enum_cls: type[enum.Enum], length: int = 16) -> Enum: + """把枚举的 value（online/offline）写入数据库，而不是成员名。""" + return Enum( + enum_cls, + native_enum=False, + length=length, + values_callable=lambda obj: [item.value for item in obj], + ) + + +class NodeStatus(str, enum.Enum): + ONLINE = "online" + OFFLINE = "offline" + + +class YieldStatus(str, enum.Enum): + SETTLED = "settled" + VOID = "void" + + +class WithdrawalStatus(str, enum.Enum): + PENDING = "pending" + COMPLETED = "completed" + REJECTED = "rejected" + FAILED = "failed" + + +class TaskStatus(str, enum.Enum): + PENDING = "pending" + COMPLETED = "completed" + FAILED = "failed" + EXPIRED = "expired" + + +class PushStatus(str, enum.Enum): + QUEUED = "queued" + PUSHED = "pushed" + FAILED = "failed" + + +class User(Base): + """用户。一个钱包地址对应一名用户，可绑定多台算力节点。""" + + __tablename__ = "users" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + wallet_address: Mapped[str] = mapped_column(String(128), unique=True, index=True) + created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) + + nodes: Mapped[list[ComputeNode]] = relationship(back_populates="user") + yields: Mapped[list[NodeYield]] = relationship(back_populates="user") + withdrawals: Mapped[list[WithdrawalLog]] = relationship(back_populates="user") + + +class ComputeNode(Base): + """算力节点。购买金额在授权周期内按天线性释放，实际到账再乘当日表现系数。""" + + __tablename__ = "compute_nodes" + __table_args__ = ( + CheckConstraint("purchase_amount > 0", name="ck_node_purchase_positive"), + CheckConstraint("license_months > 0", name="ck_node_license_months"), + Index("ix_node_status_heartbeat", "status", "last_heartbeat_at"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + sn: Mapped[str] = mapped_column(String(64), unique=True, index=True, comment="节点唯一序列号") + purchase_amount: Mapped[Decimal] = mapped_column(MONEY, comment="授权购买金额") + license_months: Mapped[int] = mapped_column(Integer, default=12) + licensed_at: Mapped[datetime] = mapped_column(UtcDateTime, comment="授权开始时间") + license_expires_at: Mapped[datetime] = mapped_column(UtcDateTime, comment="授权结束时间") + status: Mapped[NodeStatus] = mapped_column(_enum(NodeStatus), default=NodeStatus.OFFLINE, index=True) + last_heartbeat_at: Mapped[datetime | None] = mapped_column(UtcDateTime) + callback_url: Mapped[str | None] = mapped_column(String(512), comment="可选的任务推送地址") + created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) + updated_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow, onupdate=utcnow) + + user: Mapped[User] = relationship(back_populates="nodes") + yields: Mapped[list[NodeYield]] = relationship(back_populates="node") + tasks: Mapped[list[ComputeTask]] = relationship(back_populates="node") + uptime_intervals: Mapped[list[UptimeInterval]] = relationship(back_populates="node") + + +class UptimeInterval(Base): + """一段连续在线时间。心跳首次上线时打开，超时或授权到期时关闭。""" + + __tablename__ = "uptime_intervals" + __table_args__ = (Index("ix_uptime_node_window", "node_id", "started_at", "ended_at"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) + started_at: Mapped[datetime] = mapped_column(UtcDateTime) + ended_at: Mapped[datetime | None] = mapped_column(UtcDateTime) + close_reason: Mapped[str | None] = mapped_column(String(32)) + + node: Mapped[ComputeNode] = relationship(back_populates="uptime_intervals") + + +class ComputeTask(Base): + """下发给在线节点的模拟计算任务。客户端须回传 task_id 与一次性 nonce。""" + + __tablename__ = "compute_tasks" + __table_args__ = ( + Index("ix_task_node_status", "node_id", "status"), + Index("ix_task_dispatched_at", "dispatched_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, comment="task_id") + node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) + task_type: Mapped[str] = mapped_column(String(32), default="simulated_compute") + payload: Mapped[dict] = mapped_column(JSON) + status: Mapped[TaskStatus] = mapped_column(_enum(TaskStatus), default=TaskStatus.PENDING, index=True) + push_status: Mapped[PushStatus] = mapped_column(_enum(PushStatus), default=PushStatus.QUEUED) + push_error: Mapped[str | None] = mapped_column(String(255)) + dispatched_at: Mapped[datetime] = mapped_column(UtcDateTime) + deadline_at: Mapped[datetime] = mapped_column(UtcDateTime) + responded_at: Mapped[datetime | None] = mapped_column(UtcDateTime) + success: Mapped[bool | None] = mapped_column() + + node: Mapped[ComputeNode] = relationship(back_populates="tasks") + + +class NodeYield(Base): + """每日产出流水。同一节点同一天只有一条结算记录。""" + + __tablename__ = "node_yields" + __table_args__ = ( + UniqueConstraint("node_id", "yield_date", name="uq_node_yield_date"), + Index("ix_node_yield_user_date", "user_id", "yield_date"), + CheckConstraint("points_delta >= 0", name="ck_yield_points_non_negative"), + CheckConstraint("compute_output >= 0", name="ck_yield_output_non_negative"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + node_id: Mapped[int] = mapped_column(ForeignKey("compute_nodes.id"), index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + yield_date: Mapped[date] = mapped_column(index=True, comment="清算日，UTC") + compute_output: Mapped[Decimal] = mapped_column(MONEY, comment="当日有效算力产出，等于授权金额乘表现系数") + points_delta: Mapped[Decimal] = mapped_column(MONEY, comment="当日入账积分") + online_seconds: Mapped[int] = mapped_column(Integer, default=0) + eligible_seconds: Mapped[int] = mapped_column(Integer, default=0) + online_rate: Mapped[Decimal] = mapped_column(RATE, default=0) + tasks_dispatched: Mapped[int] = mapped_column(Integer, default=0) + tasks_completed: Mapped[int] = mapped_column(Integer, default=0) + task_response_rate: Mapped[Decimal] = mapped_column(RATE, default=0) + online_weight: Mapped[Decimal] = mapped_column(RATE, default=0) + task_weight: Mapped[Decimal] = mapped_column(RATE, default=0) + performance: Mapped[Decimal] = mapped_column(RATE, default=0, comment="当日表现系数，0 到 1") + cycle_days: Mapped[int] = mapped_column(Integer) + status: Mapped[YieldStatus] = mapped_column(_enum(YieldStatus), default=YieldStatus.SETTLED) + is_partial: Mapped[bool] = mapped_column(default=False, comment="结算时该 UTC 日尚未结束") + created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) + + node: Mapped[ComputeNode] = relationship(back_populates="yields") + user: Mapped[User] = relationship(back_populates="yields") + + +class WithdrawalLog(Base): + """提现申请。待处理和已完成的金额都会占用可用积分，拒绝后释放。""" + + __tablename__ = "withdrawal_logs" + __table_args__ = ( + Index("ix_withdrawal_user_status", "user_id", "status"), + CheckConstraint("amount > 0", name="ck_withdrawal_amount_positive"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) + amount: Mapped[Decimal] = mapped_column(MONEY) + asset_symbol: Mapped[str] = mapped_column(String(32), default="POINT") + tx_hash: Mapped[str | None] = mapped_column(String(128), unique=True) + status: Mapped[WithdrawalStatus] = mapped_column( + _enum(WithdrawalStatus), + default=WithdrawalStatus.PENDING, + index=True, + ) + note: Mapped[str | None] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) + updated_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow, onupdate=utcnow) + + user: Mapped[User] = relationship(back_populates="withdrawals") + + +class SchedulerLock(Base): + """定时任务锁。锁到期后其他实例才能接手，进程崩溃不会永久占住任务。""" + + __tablename__ = "scheduler_locks" + + name: Mapped[str] = mapped_column(String(64), primary_key=True) + locked_until: Mapped[datetime] = mapped_column(UtcDateTime) + owner: Mapped[str] = mapped_column(String(128)) diff --git a/app/money.py b/app/money.py new file mode 100644 index 0000000..0ecabe2 --- /dev/null +++ b/app/money.py @@ -0,0 +1,32 @@ +"""金额与比率的定点运算。积分只向下取整，避免四舍五入把释放总额做超。""" + +from __future__ import annotations + +from decimal import Decimal, ROUND_DOWN + +QUANT = Decimal("0.00000001") + + +def to_decimal(value: object) -> Decimal: + if isinstance(value, Decimal): + return value + return Decimal(str(value)) + + +def money(value: object) -> Decimal: + return to_decimal(value).quantize(QUANT, rounding=ROUND_DOWN) + + +def rate(value: object) -> Decimal: + number = to_decimal(value) + if number < 0: + number = Decimal("0") + elif number > 1: + number = Decimal("1") + return number.quantize(QUANT, rounding=ROUND_DOWN) + + +def ratio(numerator: int, denominator: int) -> Decimal: + if denominator <= 0: + return Decimal("0") + return rate(Decimal(numerator) / Decimal(denominator)) diff --git a/app/scheduler.py b/app/scheduler.py new file mode 100644 index 0000000..75d204b --- /dev/null +++ b/app/scheduler.py @@ -0,0 +1,107 @@ +"""进程内定时任务：离线巡检、随机任务下发、每日清算。""" + +from __future__ import annotations + +import logging +from datetime import timedelta + +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger +from apscheduler.triggers.interval import IntervalTrigger +from zoneinfo import ZoneInfo + +from app.config import settings +from app.database import session_scope +from app.jobs.daily_yield import run_daily_yield_cron +from app.services.heartbeat import mark_stale_nodes +from app.services.locks import acquire_lock +from app.services.tasks import dispatch_random_tasks + +logger = logging.getLogger(__name__) +UTC = ZoneInfo("UTC") +_scheduler: BackgroundScheduler | None = None + + +def start_scheduler() -> None: + global _scheduler + if settings.disable_scheduler: + logger.info("scheduler disabled") + return + if _scheduler is not None and _scheduler.running: + return + + scheduler = BackgroundScheduler(timezone=UTC, job_defaults={"coalesce": True, "max_instances": 1}) + scheduler.add_job( + _guard(_offline_job), + IntervalTrigger(minutes=settings.offline_check_minutes, timezone=UTC), + id="offline_check", + replace_existing=True, + misfire_grace_time=120, + ) + scheduler.add_job( + _guard(_dispatch_job), + IntervalTrigger(minutes=settings.task_dispatch_minutes, timezone=UTC), + id="task_dispatch", + replace_existing=True, + misfire_grace_time=120, + ) + yield_zone = ZoneInfo(settings.yield_timezone) + scheduler.add_job( + _guard(_yield_job), + CronTrigger(hour=settings.yield_hour, minute=settings.yield_minute, timezone=yield_zone), + id="daily_yield", + replace_existing=True, + misfire_grace_time=3600, + ) + scheduler.start() + _scheduler = scheduler + logger.info( + "scheduler started; daily yield at %02d:%02d %s", + settings.yield_hour, + settings.yield_minute, + settings.yield_timezone, + ) + + +def stop_scheduler() -> None: + global _scheduler + if _scheduler is not None and _scheduler.running: + _scheduler.shutdown(wait=False) + _scheduler = None + + +def _guard(func): + def wrapper() -> None: + try: + func() + except Exception: + logger.exception("scheduled job failed: %s", func.__name__) + + return wrapper + + +def _offline_job() -> None: + ttl = timedelta(seconds=max(30, settings.offline_check_minutes * 60 - 30)) + if not acquire_lock("offline_check", ttl): + return + with session_scope() as session: + changed = mark_stale_nodes(session) + if changed: + logger.info("marked offline: %s", ",".join(changed)) + + +def _dispatch_job() -> None: + ttl = timedelta(seconds=max(30, settings.task_dispatch_minutes * 60 - 30)) + if not acquire_lock("task_dispatch", ttl): + return + with session_scope() as session: + tasks = dispatch_random_tasks(session) + if tasks: + logger.info("dispatched %s tasks", len(tasks)) + + +def _yield_job() -> None: + if not acquire_lock("daily_yield", timedelta(hours=2)): + logger.info("daily yield skipped, lock held") + return + run_daily_yield_cron() diff --git a/app/schemas.py b/app/schemas.py new file mode 100644 index 0000000..1838630 --- /dev/null +++ b/app/schemas.py @@ -0,0 +1,156 @@ +"""接口请求与响应。金额在 JSON 里用字符串，避免浮点误差。""" + +from __future__ import annotations + +from datetime import date, datetime +from decimal import Decimal +from typing import Annotated + +from pydantic import BaseModel, ConfigDict, Field, PlainSerializer + +from app.models import NodeStatus, TaskStatus, WithdrawalStatus, YieldStatus + +SN = Annotated[str, Field(min_length=4, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")] +WALLET = Annotated[str, Field(min_length=4, max_length=128, pattern=r"^[A-Za-z0-9:_-]+$")] +MoneyText = Annotated[Decimal, PlainSerializer(lambda value: format(value, "f"), return_type=str)] + + +class APIModel(BaseModel): + model_config = ConfigDict(from_attributes=True) + + +class UserCreate(APIModel): + wallet_address: WALLET + + +class UserOut(APIModel): + id: int + wallet_address: str + created_at: datetime + + +class NodeCreate(APIModel): + wallet_address: WALLET + sn: SN + purchase_amount: Decimal = Field(gt=0, max_digits=24, decimal_places=8) + callback_url: str | None = None + + +class NodeOut(APIModel): + id: int + user_id: int + sn: str + purchase_amount: MoneyText + license_months: int + licensed_at: datetime + license_expires_at: datetime + status: NodeStatus + last_heartbeat_at: datetime | None + callback_url: str | None + + +class HeartbeatIn(APIModel): + sn: SN + + +class HeartbeatOut(APIModel): + sn: str + status: NodeStatus + last_heartbeat_at: datetime + server_time: datetime + + +class TaskOut(APIModel): + task_id: str + task_type: str + payload: dict + status: TaskStatus + dispatched_at: datetime + deadline_at: datetime + + +class TaskReportIn(APIModel): + sn: SN + task_id: str = Field(min_length=36, max_length=36) + success: bool + nonce: str = Field(min_length=8, max_length=128) + + +class ContributionOut(APIModel): + sn: str + yield_date: date + eligible: bool + online_seconds: int + eligible_seconds: int + online_rate: MoneyText + tasks_dispatched: int + tasks_completed: int + task_response_rate: MoneyText + online_weight: MoneyText + task_weight: MoneyText + performance: MoneyText + compute_output: MoneyText + points_delta: MoneyText + cycle_days: int + + +class YieldOut(APIModel): + id: int + node_id: int + user_id: int + yield_date: date + compute_output: MoneyText + points_delta: MoneyText + online_seconds: int + eligible_seconds: int + online_rate: MoneyText + tasks_dispatched: int + tasks_completed: int + task_response_rate: MoneyText + online_weight: MoneyText + task_weight: MoneyText + performance: MoneyText + cycle_days: int + status: YieldStatus + is_partial: bool + created_at: datetime + + +class SettleOut(APIModel): + yield_date: date + created: int + skipped: int + total_points: MoneyText + is_partial: bool + + +class WithdrawalCreate(APIModel): + wallet_address: WALLET + amount: Decimal = Field(gt=0, max_digits=24, decimal_places=8) + asset_symbol: str = Field(default="POINT", min_length=1, max_length=32) + + +class WithdrawalConfirm(APIModel): + tx_hash: str = Field(min_length=6, max_length=128, pattern=r"^[A-Za-z0-9]+$") + + +class WithdrawalReject(APIModel): + note: str | None = Field(default=None, max_length=500) + + +class WithdrawalOut(APIModel): + id: int + user_id: int + amount: MoneyText + asset_symbol: str + tx_hash: str | None + status: WithdrawalStatus + note: str | None + available_after: MoneyText | None = None + created_at: datetime + updated_at: datetime + + +class BalanceOut(APIModel): + wallet_address: str + available_points: MoneyText diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..1672939 --- /dev/null +++ b/app/services/__init__.py @@ -0,0 +1 @@ +"""业务服务。""" diff --git a/app/services/common.py b/app/services/common.py new file mode 100644 index 0000000..2f8676e --- /dev/null +++ b/app/services/common.py @@ -0,0 +1,21 @@ +"""按序列号或钱包地址读取主体，供各业务模块共用。""" + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.errors import AppError +from app.models import ComputeNode, User + + +def get_user_by_wallet(session: Session, wallet_address: str) -> User: + user = session.scalar(select(User).where(User.wallet_address == wallet_address)) + if user is None: + raise AppError("用户不存在", 404) + return user + + +def get_node_by_sn(session: Session, sn: str) -> ComputeNode: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == sn)) + if node is None: + raise AppError(f"节点不存在: {sn}", 404) + return node diff --git a/app/services/heartbeat.py b/app/services/heartbeat.py new file mode 100644 index 0000000..b7f6fc1 --- /dev/null +++ b/app/services/heartbeat.py @@ -0,0 +1,78 @@ +"""心跳登记与离线巡检。""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.clock import utcnow +from app.config import settings +from app.errors import AppError +from app.models import ComputeNode, NodeStatus, UptimeInterval +from app.services.common import get_node_by_sn + + +def record_heartbeat(session: Session, sn: str, now: datetime | None = None) -> ComputeNode: + """刷新最后活跃时间。离线节点收到心跳后重新打开一段在线区间。""" + moment = now or utcnow() + node = get_node_by_sn(session, sn.strip()) + if node.license_expires_at <= moment: + raise AppError("节点授权已到期", 403) + + node.last_heartbeat_at = moment + node.updated_at = moment + if node.status != NodeStatus.ONLINE: + node.status = NodeStatus.ONLINE + session.add(UptimeInterval(node_id=node.id, started_at=moment)) + else: + _ensure_open_interval(session, node.id, moment) + session.flush() + return node + + +def mark_stale_nodes(session: Session, now: datetime | None = None) -> list[str]: + """把超过心跳时限或授权已到期的在线节点置为 offline，并关闭在线区间。""" + moment = now or utcnow() + cutoff = moment - timedelta(minutes=settings.heartbeat_timeout_minutes) + nodes = session.scalars(select(ComputeNode).where(ComputeNode.status == NodeStatus.ONLINE)).all() + changed: list[str] = [] + for node in nodes: + reasons: list[tuple[datetime, str]] = [] + if node.last_heartbeat_at is None or node.last_heartbeat_at < cutoff: + reasons.append((node.last_heartbeat_at or moment, "heartbeat_timeout")) + if node.license_expires_at <= moment: + reasons.append((node.license_expires_at, "license_expired")) + if not reasons: + continue + ended_at, reason = min(reasons, key=lambda item: item[0]) + _close_open_intervals(session, node.id, ended_at, reason) + node.status = NodeStatus.OFFLINE + node.updated_at = moment + changed.append(node.sn) + session.flush() + return changed + + +def _ensure_open_interval(session: Session, node_id: int, started_at: datetime) -> None: + opened = session.scalar( + select(UptimeInterval).where( + UptimeInterval.node_id == node_id, + UptimeInterval.ended_at.is_(None), + ) + ) + if opened is None: + session.add(UptimeInterval(node_id=node_id, started_at=started_at)) + + +def _close_open_intervals(session: Session, node_id: int, ended_at: datetime, reason: str) -> None: + rows = session.scalars( + select(UptimeInterval).where( + UptimeInterval.node_id == node_id, + UptimeInterval.ended_at.is_(None), + ) + ).all() + for row in rows: + row.ended_at = ended_at if ended_at >= row.started_at else row.started_at + row.close_reason = reason diff --git a/app/services/locks.py b/app/services/locks.py new file mode 100644 index 0000000..767f556 --- /dev/null +++ b/app/services/locks.py @@ -0,0 +1,41 @@ +"""多实例定时任务互斥。锁有过期时间，持有进程异常退出后其他实例仍能继续。""" + +from __future__ import annotations + +import os +import socket +from datetime import timedelta + +from sqlalchemy import update +from sqlalchemy.exc import IntegrityError + +from app.clock import utcnow +from app.database import session_scope +from app.models import SchedulerLock + +OWNER = f"{socket.gethostname()}:{os.getpid()}" + + +def acquire_lock(name: str, ttl: timedelta) -> bool: + now = utcnow() + until = now + ttl + try: + with session_scope() as session: + lock = session.get(SchedulerLock, name) + if lock is None: + session.add(SchedulerLock(name=name, locked_until=until, owner=OWNER)) + try: + session.flush() + except IntegrityError: + session.rollback() + return False + return True + updated = session.execute( + update(SchedulerLock) + .where(SchedulerLock.name == name) + .where(SchedulerLock.locked_until <= now) + .values(locked_until=until, owner=OWNER) + ) + return updated.rowcount == 1 + except IntegrityError: + return False diff --git a/app/services/registry.py b/app/services/registry.py new file mode 100644 index 0000000..8f24cc4 --- /dev/null +++ b/app/services/registry.py @@ -0,0 +1,74 @@ +"""用户与节点登记。""" + +from __future__ import annotations + +from decimal import Decimal +from urllib.parse import urlparse + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.clock import add_months, utcnow +from app.config import settings +from app.errors import AppError +from app.models import ComputeNode, NodeStatus, User +from app.money import money +from app.services.common import get_user_by_wallet + + +def register_user(session: Session, wallet_address: str) -> User: + wallet = wallet_address.strip() + if session.scalar(select(User).where(User.wallet_address == wallet)) is not None: + raise AppError("钱包地址已注册", 409) + user = User(wallet_address=wallet, created_at=utcnow()) + session.add(user) + try: + session.flush() + except IntegrityError as exc: + session.rollback() + raise AppError("钱包地址已注册", 409) from exc + return user + + +def register_node( + session: Session, + wallet_address: str, + sn: str, + purchase_amount: Decimal, + callback_url: str | None = None, +) -> ComputeNode: + user = get_user_by_wallet(session, wallet_address.strip()) + serial = sn.strip() + if session.scalar(select(ComputeNode).where(ComputeNode.sn == serial)) is not None: + raise AppError("节点序列号已存在", 409) + callback = _normalize_callback(callback_url) + started = utcnow() + node = ComputeNode( + user_id=user.id, + sn=serial, + purchase_amount=money(purchase_amount), + license_months=settings.license_months, + licensed_at=started, + license_expires_at=add_months(started, settings.license_months), + status=NodeStatus.OFFLINE, + callback_url=callback, + created_at=started, + updated_at=started, + ) + session.add(node) + try: + session.flush() + except IntegrityError as exc: + session.rollback() + raise AppError("节点序列号已存在", 409) from exc + return node + + +def _normalize_callback(callback_url: str | None) -> str | None: + if callback_url is None or callback_url.strip() == "": + return None + parsed = urlparse(callback_url.strip()) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise AppError("callback_url 必须是 http 或 https 地址") + return callback_url.strip() diff --git a/app/services/tasks.py b/app/services/tasks.py new file mode 100644 index 0000000..87f55dc --- /dev/null +++ b/app/services/tasks.py @@ -0,0 +1,183 @@ +"""随机任务下发与回传校验。 + +节点不暴露回调地址时，任务留在待拉取队列，客户端用 +GET /api/v1/node/{sn}/tasks 领取。配置了 callback_url 时，调度器会额外 POST 一次。 +回传必须带上任务里的 nonce，避免未收到任务的客户端直接上报成功。 +""" + +from __future__ import annotations + +import hmac +import json +import logging +import random +import secrets +import uuid +from datetime import datetime, timedelta +from urllib.error import URLError +from urllib.request import Request, urlopen + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.clock import utcnow +from app.config import settings +from app.errors import AppError +from app.models import ComputeNode, ComputeTask, NodeStatus, PushStatus, TaskStatus +from app.services.common import get_node_by_sn + +logger = logging.getLogger(__name__) + + +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() + ) + candidates = [ + node + for node in session.scalars( + select(ComputeNode).where( + ComputeNode.status == NodeStatus.ONLINE, + ComputeNode.license_expires_at > moment, + ) + ).all() + if node.id not in pending_node_ids + ] + if not candidates: + return [] + + batch = min(len(candidates), settings.dispatch_batch_size) + chosen = random.sample(candidates, batch) + deadline = moment + timedelta(minutes=settings.task_deadline_minutes) + tasks: list[ComputeTask] = [] + for node in chosen: + task = ComputeTask( + id=str(uuid.uuid4()), + node_id=node.id, + task_type="simulated_compute", + payload={ + "op": "simulated_compute", + "nonce": secrets.token_hex(16), + "seed": secrets.token_hex(8), + }, + 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(chosen, tasks): + _push_task(node, task) + session.flush() + return tasks + + +def list_pending_tasks(session: Session, sn: str, now: datetime | None = None) -> list[ComputeTask]: + moment = now or utcnow() + node = get_node_by_sn(session, sn.strip()) + tasks = session.scalars( + select(ComputeTask) + .where(ComputeTask.node_id == node.id, ComputeTask.status == TaskStatus.PENDING) + .order_by(ComputeTask.dispatched_at) + ).all() + visible: list[ComputeTask] = [] + for task in tasks: + if task.deadline_at < moment: + task.status = TaskStatus.EXPIRED + continue + visible.append(task) + session.flush() + return visible + + +def report_task( + session: Session, + sn: str, + task_id: str, + success: bool, + nonce: str, + now: datetime | None = None, +) -> ComputeTask: + """校验任务归属、时限和 nonce。只有校验通过且 success 为真才记为有效响应。""" + moment = now or utcnow() + node = get_node_by_sn(session, sn.strip()) + task = session.get(ComputeTask, task_id) + if task is None: + raise AppError("任务不存在", 404) + if task.node_id != node.id: + raise AppError("任务不属于该节点", 403) + if task.status != TaskStatus.PENDING: + raise AppError("任务已结束，不能重复回传", 409) + if task.deadline_at < moment: + task.status = TaskStatus.EXPIRED + session.flush() + raise AppError("任务已超过回传时限", 409) + + expected = str((task.payload or {}).get("nonce", "")) + if not _nonce_matches(nonce, expected): + raise AppError("任务响应校验失败") + + task.responded_at = moment + task.success = success + task.status = TaskStatus.COMPLETED if success else TaskStatus.FAILED + session.flush() + return task + + +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.deadline_at < moment, + ) + ).all() + for task in tasks: + task.status = TaskStatus.EXPIRED + if tasks: + session.flush() + return len(tasks) + + +def _nonce_matches(provided: str, expected: str) -> bool: + if not expected or len(provided) != len(expected): + return False + return hmac.compare_digest(provided, expected) + + +def _push_task(node: ComputeNode, task: ComputeTask) -> None: + if not node.callback_url: + task.push_status = PushStatus.QUEUED + return + body = json.dumps( + { + "task_id": task.id, + "sn": node.sn, + "task_type": task.task_type, + "payload": task.payload, + "deadline_at": task.deadline_at.isoformat(), + } + ).encode() + request = Request( + node.callback_url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urlopen(request, timeout=2) as response: + if 200 <= response.status < 300: + task.push_status = PushStatus.PUSHED + task.push_error = None + return + task.push_status = PushStatus.FAILED + task.push_error = f"HTTP {response.status}" + except (URLError, TimeoutError, OSError) as exc: + task.push_status = PushStatus.FAILED + task.push_error = str(exc)[:255] + logger.warning("task push failed sn=%s task=%s error=%s", node.sn, task.id, exc) diff --git a/app/services/withdrawals.py b/app/services/withdrawals.py new file mode 100644 index 0000000..a6a3df2 --- /dev/null +++ b/app/services/withdrawals.py @@ -0,0 +1,99 @@ +"""提现申请。可用余额 = 已入账积分 - 待处理提现 - 已完成提现。""" + +from __future__ import annotations + +from decimal import Decimal + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.clock import utcnow +from app.errors import AppError +from app.models import NodeYield, WithdrawalLog, WithdrawalStatus, YieldStatus +from app.money import money, to_decimal +from app.services.common import get_user_by_wallet + + +def available_balance(session: Session, user_id: int) -> Decimal: + credited_rows = session.scalars( + select(NodeYield.points_delta).where( + NodeYield.user_id == user_id, + NodeYield.status == YieldStatus.SETTLED, + ) + ).all() + reserved_rows = session.scalars( + select(WithdrawalLog.amount).where( + WithdrawalLog.user_id == user_id, + WithdrawalLog.status.in_([WithdrawalStatus.PENDING, WithdrawalStatus.COMPLETED]), + ) + ).all() + credited = sum((to_decimal(item) for item in credited_rows), Decimal("0")) + reserved = sum((to_decimal(item) for item in reserved_rows), Decimal("0")) + balance = credited - reserved + if balance <= 0: + return Decimal("0") + return money(balance) + + +def request_withdrawal( + session: Session, + wallet_address: str, + amount: Decimal, + asset_symbol: str = "POINT", +) -> tuple[WithdrawalLog, Decimal]: + user = get_user_by_wallet(session, wallet_address.strip()) + value = money(amount) + if value <= 0: + raise AppError("提现金额必须大于 0") + balance = available_balance(session, user.id) + if value > balance: + raise AppError("可用积分不足") + row = WithdrawalLog( + user_id=user.id, + amount=value, + asset_symbol=asset_symbol.strip().upper() or "POINT", + status=WithdrawalStatus.PENDING, + created_at=utcnow(), + updated_at=utcnow(), + ) + session.add(row) + session.flush() + return row, money(balance - value) + + +def confirm_withdrawal(session: Session, withdrawal_id: int, tx_hash: str) -> WithdrawalLog: + row = _get_withdrawal(session, withdrawal_id) + if row.status != WithdrawalStatus.PENDING: + raise AppError("仅待处理的提现可以确认", 409) + digest = tx_hash.strip() + occupied = session.scalar(select(WithdrawalLog.id).where(WithdrawalLog.tx_hash == digest)) + if occupied is not None: + raise AppError("交易哈希已存在", 409) + row.tx_hash = digest + row.status = WithdrawalStatus.COMPLETED + row.updated_at = utcnow() + try: + session.flush() + except IntegrityError as exc: + session.rollback() + raise AppError("交易哈希已存在", 409) from exc + return row + + +def reject_withdrawal(session: Session, withdrawal_id: int, note: str | None = None) -> WithdrawalLog: + row = _get_withdrawal(session, withdrawal_id) + if row.status != WithdrawalStatus.PENDING: + raise AppError("仅待处理的提现可以拒绝", 409) + row.status = WithdrawalStatus.REJECTED + row.note = note + row.updated_at = utcnow() + session.flush() + return row + + +def _get_withdrawal(session: Session, withdrawal_id: int) -> WithdrawalLog: + row = session.get(WithdrawalLog, withdrawal_id) + if row is None: + raise AppError("提现记录不存在", 404) + return row diff --git a/app/services/yield_engine.py b/app/services/yield_engine.py new file mode 100644 index 0000000..fea8dea --- /dev/null +++ b/app/services/yield_engine.py @@ -0,0 +1,236 @@ +"""每日产出清算。 + +释放规则： +1. 授权金额在整个授权天数内线性释放。授权天数取 license 起止日期之差。 +2. 在线率 = 当日授权窗口内的在线秒数 / 当日授权有效秒数。 + 尚未关闭的在线区间，若最后心跳已超过超时时间，只计算到最后一次心跳。 +3. 任务响应率 = 校验通过且执行成功的任务数 / 当日内下发任务数。 +4. 当天没有下发任务时，表现系数等于在线率，避免调度器故障误伤节点。 + 有任务时，表现系数 = 在线权重 × 在线率 + 任务权重 × 响应率。 +5. 算力产出 = 授权金额 × 表现系数。 + 当日积分 = 算力产出 / 授权天数。 +6. 单节点累计积分不超过授权金额。已存在的当日流水不会被覆盖。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from decimal import Decimal + +from sqlalchemy import or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.clock import day_bounds, utcnow +from app.config import settings +from app.errors import AppError +from app.models import ComputeNode, ComputeTask, NodeYield, TaskStatus, UptimeInterval, YieldStatus +from app.money import money, rate, ratio, to_decimal +from app.services.common import get_node_by_sn +from app.services.tasks import expire_overdue_tasks + + +@dataclass +class YieldComputation: + eligible: bool + yield_date: date + online_seconds: int = 0 + eligible_seconds: int = 0 + online_rate: Decimal = Decimal("0") + tasks_dispatched: int = 0 + tasks_completed: int = 0 + task_response_rate: Decimal = Decimal("0") + online_weight: Decimal = Decimal("0") + task_weight: Decimal = Decimal("0") + performance: Decimal = Decimal("0") + compute_output: Decimal = Decimal("0") + points_delta: Decimal = Decimal("0") + cycle_days: int = 0 + + +@dataclass +class SettleSummary: + yield_date: date + created: int + skipped: int + total_points: Decimal + is_partial: bool + + +def compute_node_yield( + session: Session, + node: ComputeNode, + yield_date: date, + now: datetime | None = None, +) -> YieldComputation: + moment = now or utcnow() + day_start, day_end = day_bounds(yield_date) + window_start = max(day_start, node.licensed_at) + window_end = min(day_end, node.license_expires_at) + cycle_days = (node.license_expires_at.date() - node.licensed_at.date()).days + if window_end <= window_start or cycle_days <= 0: + return YieldComputation(eligible=False, yield_date=yield_date, cycle_days=max(cycle_days, 0)) + + eligible_seconds = int((window_end - window_start).total_seconds()) + online_seconds = min(_online_seconds(session, node, window_start, window_end, moment), eligible_seconds) + online_rate = ratio(online_seconds, eligible_seconds) + + tasks = session.scalars( + select(ComputeTask).where( + ComputeTask.node_id == node.id, + ComputeTask.dispatched_at >= window_start, + ComputeTask.dispatched_at < window_end, + ) + ).all() + tasks_dispatched = len(tasks) + tasks_completed = sum(1 for task in tasks if task.status == TaskStatus.COMPLETED) + response_rate = ratio(tasks_completed, tasks_dispatched) if tasks_dispatched else Decimal("0") + + if tasks_dispatched == 0: + applied_online = Decimal("1") + applied_task = Decimal("0") + else: + applied_online = to_decimal(settings.online_weight) + applied_task = to_decimal(settings.task_weight) + performance = rate(applied_online * online_rate + applied_task * response_rate) + compute_output = money(to_decimal(node.purchase_amount) * performance) + points_delta = money(compute_output / Decimal(cycle_days)) + points_delta = min(points_delta, _remaining_points(session, node)) + + return YieldComputation( + eligible=True, + yield_date=yield_date, + online_seconds=online_seconds, + eligible_seconds=eligible_seconds, + online_rate=online_rate, + tasks_dispatched=tasks_dispatched, + tasks_completed=tasks_completed, + task_response_rate=response_rate, + online_weight=money(applied_online), + task_weight=money(applied_task), + performance=performance, + compute_output=compute_output, + points_delta=points_delta, + cycle_days=cycle_days, + ) + + +def settle_yield_date(session: Session, yield_date: date, now: datetime | None = None) -> SettleSummary: + """清算指定 UTC 日。每个节点当天已有流水时跳过，保证脚本重复执行不会重复入账。""" + moment = now or utcnow() + if yield_date > moment.date(): + raise AppError("不能清算未来日期") + expire_overdue_tasks(session, moment) + day_start, day_end = day_bounds(yield_date) + nodes = session.scalars( + select(ComputeNode) + .where( + ComputeNode.licensed_at < day_end, + ComputeNode.license_expires_at > day_start, + ) + .order_by(ComputeNode.id) + ).all() + created = 0 + skipped = 0 + total = Decimal("0") + partial = yield_date == moment.date() + for node in nodes: + exists = session.scalar( + select(NodeYield.id).where(NodeYield.node_id == node.id, NodeYield.yield_date == yield_date) + ) + if exists is not None: + skipped += 1 + continue + computed = compute_node_yield(session, node, yield_date, moment) + if not computed.eligible: + continue + try: + with session.begin_nested(): + session.add(_to_row(node, computed, partial)) + except IntegrityError: + skipped += 1 + continue + created += 1 + total += computed.points_delta + return SettleSummary( + yield_date=yield_date, + created=created, + skipped=skipped, + total_points=money(total), + is_partial=partial, + ) + + +def list_node_yields(session: Session, sn: str) -> list[NodeYield]: + node = get_node_by_sn(session, sn.strip()) + return list( + session.scalars( + select(NodeYield).where(NodeYield.node_id == node.id).order_by(NodeYield.yield_date.desc()) + ).all() + ) + + +def _online_seconds( + session: Session, + node: ComputeNode, + window_start: datetime, + window_end: datetime, + now: datetime, +) -> int: + intervals = session.scalars( + select(UptimeInterval).where( + UptimeInterval.node_id == node.id, + UptimeInterval.started_at < window_end, + or_(UptimeInterval.ended_at.is_(None), UptimeInterval.ended_at > window_start), + ) + ).all() + total = 0 + for interval in intervals: + start = max(interval.started_at, window_start) + end = min(_proven_end(interval, node, now), window_end) + if end > start: + total += int((end - start).total_seconds()) + return total + + +def _proven_end(interval: UptimeInterval, node: ComputeNode, now: datetime) -> datetime: + """开放区间只能证明到最后一次心跳；心跳仍在超时窗口内才延续到当前时间。""" + if interval.ended_at is not None: + return interval.ended_at + last_seen = node.last_heartbeat_at or interval.started_at + if last_seen + timedelta(minutes=settings.heartbeat_timeout_minutes) < now: + return last_seen + return now + + +def _remaining_points(session: Session, node: ComputeNode) -> Decimal: + paid_rows = session.scalars(select(NodeYield.points_delta).where(NodeYield.node_id == node.id)).all() + paid = sum((to_decimal(item) for item in paid_rows), Decimal("0")) + remaining = to_decimal(node.purchase_amount) - paid + if remaining <= 0: + return Decimal("0") + return money(remaining) + + +def _to_row(node: ComputeNode, computed: YieldComputation, partial: bool) -> NodeYield: + return NodeYield( + node_id=node.id, + user_id=node.user_id, + yield_date=computed.yield_date, + compute_output=computed.compute_output, + points_delta=computed.points_delta, + online_seconds=computed.online_seconds, + eligible_seconds=computed.eligible_seconds, + online_rate=computed.online_rate, + tasks_dispatched=computed.tasks_dispatched, + tasks_completed=computed.tasks_completed, + task_response_rate=computed.task_response_rate, + online_weight=computed.online_weight, + task_weight=computed.task_weight, + performance=computed.performance, + cycle_days=computed.cycle_days, + status=YieldStatus.SETTLED, + is_partial=partial, + created_at=utcnow(), + ) diff --git a/app/settle.py b/app/settle.py new file mode 100644 index 0000000..b5669a1 --- /dev/null +++ b/app/settle.py @@ -0,0 +1,35 @@ +"""独立的每日清算脚本。服务进程里的定时任务也会调用同一套结算函数。 + +示例： + python -m app.settle + python -m app.settle --date 2026-10-09 +""" + +from __future__ import annotations + +import argparse +import logging +from datetime import date + +import app.models # noqa: F401 +from app.database import Base, engine +from app.jobs.daily_yield import run_daily_yield_cron + + +def main(argv: list[str] | None = None) -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + parser = argparse.ArgumentParser(description="按 UTC 日清算节点积分产出") + parser.add_argument("--date", dest="yield_date", help="清算日期 YYYY-MM-DD，默认上一个已结束的 UTC 日") + args = parser.parse_args(argv) + target = date.fromisoformat(args.yield_date) if args.yield_date else None + + Base.metadata.create_all(bind=engine) + summary = run_daily_yield_cron(yield_date=target) + print( + f"yield_date={summary.yield_date} created={summary.created} " + f"skipped={summary.skipped} total_points={format(summary.total_points, 'f')} partial={summary.is_partial}" + ) + + +if __name__ == "__main__": + main() diff --git a/deploy/ubuntu22.sh b/deploy/ubuntu22.sh new file mode 100644 index 0000000..c9e27d8 --- /dev/null +++ b/deploy/ubuntu22.sh @@ -0,0 +1,264 @@ +#!/usr/bin/env bash +# HyperNode-AI 交互式部署脚本，适用于 Ubuntu 22.04。 +# 不安装 Nginx。反代由你自己配置，服务监听 0.0.0.0:33333。 +# +# sudo bash deploy/ubuntu22.sh + +set -euo pipefail + +APP_USER="hypernode" +APP_DIR="/opt/hypernode" +DATA_DIR="/var/lib/hypernode" +ENV_FILE="${APP_DIR}/.env" +SERVICE_NAME="hypernode" +BIND_HOST="0.0.0.0" +BIND_PORT="33333" +SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +if [[ "${EUID}" -ne 0 ]]; then + echo "请用 root 运行：sudo bash deploy/ubuntu22.sh" >&2 + exit 1 +fi + +if [[ ! -t 0 ]]; then + echo "请在交互式终端中运行本脚本。" >&2 + exit 1 +fi + +if [[ ! -r /etc/os-release ]]; then + echo "无法识别操作系统，这个脚本用于 Ubuntu 22.04。" >&2 + exit 1 +fi +# shellcheck disable=SC1091 +source /etc/os-release +if [[ "${ID:-}" != "ubuntu" ]]; then + echo "当前系统是 ${ID:-unknown}，这个脚本用于 Ubuntu 22.04。" >&2 + exit 1 +fi +if [[ "${VERSION_ID:-}" != "22.04" ]]; then + echo "警告：当前是 Ubuntu ${VERSION_ID:-unknown}，脚本按 22.04 编写，将继续。" >&2 +fi + +if [[ ! -f "${SOURCE_DIR}/requirements.txt" || ! -f "${SOURCE_DIR}/app/main.py" ]]; then + echo "未在 ${SOURCE_DIR} 找到项目文件。" >&2 + exit 1 +fi + +log() { + printf '\n==> %s\n' "$*" +} + +pause() { + read -r -p "按回车返回菜单..." _ || exit 0 +} + +sync_code() { + mkdir -p "${APP_DIR}" "${DATA_DIR}" + if [[ "$(realpath "${SOURCE_DIR}")" != "$(realpath "${APP_DIR}")" ]]; then + rsync -a --delete \ + --exclude '.venv/' \ + --exclude '.env' \ + --exclude '.git/' \ + --exclude '__pycache__/' \ + --exclude '*.db' \ + --exclude '.pytest_cache/' \ + "${SOURCE_DIR}/" "${APP_DIR}/" + fi +} + +write_env() { + if [[ -f "${ENV_FILE}" ]]; then + log "保留已有配置 ${ENV_FILE}" + return + fi + log "写入 ${ENV_FILE}" + cat >"${ENV_FILE}" <"/etc/systemd/system/${SERVICE_NAME}.service" </dev/null; then + ready=1 + break + fi + sleep 1 + done + if [[ "${ready}" -ne 1 ]]; then + echo "服务未在 10 秒内就绪。最近日志：" >&2 + journalctl -u "${SERVICE_NAME}.service" -n 50 --no-pager >&2 || true + return 1 + fi +} + +ensure_user() { + if ! id -u "${APP_USER}" >/dev/null 2>&1; then + useradd --system --home-dir "${APP_DIR}" --shell /usr/sbin/nologin "${APP_USER}" + fi +} + +install_packages() { + export DEBIAN_FRONTEND=noninteractive + apt-get update + apt-get install -y python3 python3-venv python3-pip rsync ca-certificates curl + python3 -c 'import sys; raise SystemExit(0 if sys.version_info >= (3, 10) else 1)' +} + +cmd_deploy() { + log "一键部署" + install_packages + ensure_user + sync_code + write_env + fix_permissions + install_python_deps + fix_permissions + write_service + systemctl enable "${SERVICE_NAME}.service" + systemctl restart "${SERVICE_NAME}.service" + wait_until_ready + log "部署完成" + echo "监听地址：${BIND_HOST}:${BIND_PORT}" + echo "健康检查：curl http://127.0.0.1:${BIND_PORT}/health" + echo "接口文档：http://<服务器 IP>:${BIND_PORT}/docs" + echo "程序目录：${APP_DIR}" + echo "数据库：${DATA_DIR}/hypernode.db" + echo "日志：journalctl -u ${SERVICE_NAME}.service -f" + echo "反代请自行配置，上游地址使用 127.0.0.1:${BIND_PORT}" +} + +cmd_update() { + if [[ ! -f "/etc/systemd/system/${SERVICE_NAME}.service" ]]; then + echo "尚未部署。请先选择「一键部署」。" >&2 + return 1 + fi + log "更新程序" + install_packages + sync_code + install_python_deps + fix_permissions + write_service + systemctl restart "${SERVICE_NAME}.service" + wait_until_ready + log "更新完成，配置和数据库已保留" +} + +cmd_uninstall() { + local answer + echo "即将停止服务并删除 ${APP_DIR} 与 systemd 单元。" + read -r -p "确认卸载请输入 yes：" answer || exit 0 + if [[ "${answer}" != "yes" ]]; then + echo "已取消卸载。" + return 0 + fi + + log "卸载服务" + systemctl stop "${SERVICE_NAME}.service" || true + systemctl disable "${SERVICE_NAME}.service" || true + rm -f "/etc/systemd/system/${SERVICE_NAME}.service" + systemctl daemon-reload + rm -rf "${APP_DIR}" + + read -r -p "同时删除数据库 ${DATA_DIR} ？输入 yes 删除，其他键保留：" answer || exit 0 + if [[ "${answer}" == "yes" ]]; then + rm -rf "${DATA_DIR}" + echo "数据库已删除。" + else + echo "数据库已保留在 ${DATA_DIR}" + fi + + if id -u "${APP_USER}" >/dev/null 2>&1; then + userdel "${APP_USER}" || echo "系统用户 ${APP_USER} 未删除，可稍后执行 userdel ${APP_USER}" + fi + log "卸载完成" +} + +show_menu() { + cat <=0.115 +uvicorn>=0.32 +sqlalchemy>=2.0.30 +apscheduler>=3.10.4,<4 +pydantic>=2.6 +pydantic-settings>=2.2 +httpx>=0.27 +tzdata>=2024.1 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_core_flow.py b/tests/test_core_flow.py new file mode 100644 index 0000000..8331f29 --- /dev/null +++ b/tests/test_core_flow.py @@ -0,0 +1,294 @@ +"""核心链路回归：心跳超时、任务校验、产出公式、提现余额。""" + +from __future__ import annotations + +import os +import tempfile +import unittest +import uuid +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from pathlib import Path + +_db_path = Path(tempfile.gettempdir()) / "depin_core_flow.db" +os.environ["DATABASE_URL"] = "sqlite:///" + _db_path.as_posix() +os.environ["DISABLE_SCHEDULER"] = "1" + +from fastapi.testclient import TestClient +from sqlalchemy import select + +from app.clock import utcnow +from app.database import Base, SessionLocal, engine +from app.main import app +from app.jobs.daily_yield import run_daily_yield_cron +from app.models import ( + ComputeNode, + ComputeTask, + NodeStatus, + NodeYield, + PushStatus, + TaskStatus, + UptimeInterval, + YieldStatus, +) +from app.services.heartbeat import mark_stale_nodes +from app.services.tasks import dispatch_random_tasks + + +class CoreFlowTest(unittest.TestCase): + def setUp(self) -> None: + Base.metadata.drop_all(bind=engine) + Base.metadata.create_all(bind=engine) + self.client_cm = TestClient(app) + self.client = self.client_cm.__enter__() + + def tearDown(self) -> None: + self.client_cm.__exit__(None, None, None) + + def test_heartbeat_goes_offline_after_timeout(self) -> None: + self._register("0xabc", "NODE-A", "100") + beat = self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-A"}) + self.assertEqual(beat.status_code, 200, beat.text) + self.assertEqual(beat.json()["status"], "online") + + recent = utcnow() - timedelta(minutes=1) + self._shift_presence("NODE-A", recent) + with SessionLocal() as session: + mark_stale_nodes(session, utcnow()) + session.commit() + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-A")) + self.assertEqual(node.status, NodeStatus.ONLINE) + + stale_at = utcnow() - timedelta(minutes=30) + self._shift_presence("NODE-A", stale_at) + with SessionLocal() as session: + changed = mark_stale_nodes(session, utcnow()) + session.commit() + self.assertEqual(changed, ["NODE-A"]) + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-A")) + interval = session.scalar(select(UptimeInterval).where(UptimeInterval.node_id == node.id)) + self.assertEqual(node.status, NodeStatus.OFFLINE) + self.assertEqual(interval.ended_at, node.last_heartbeat_at) + self.assertAlmostEqual((utcnow() - interval.ended_at).total_seconds(), 1800, delta=5) + self.assertEqual(interval.close_reason, "heartbeat_timeout") + + expired = self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-A"}) + self.assertEqual(expired.status_code, 200, expired.text) + + with SessionLocal() as session: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-A")) + node.license_expires_at = utcnow() - timedelta(minutes=1) + session.commit() + denied = self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-A"}) + self.assertEqual(denied.status_code, 403, denied.text) + + def test_task_report_requires_nonce(self) -> None: + self._register("0xdef", "NODE-T", "50") + self.client.post("/api/v1/node/heartbeat", json={"sn": "NODE-T"}) + with SessionLocal() as session: + tasks = dispatch_random_tasks(session) + session.commit() + task_id = tasks[0].id + nonce = tasks[0].payload["nonce"] + + pending = self.client.get("/api/v1/node/NODE-T/tasks") + self.assertEqual(pending.status_code, 200, pending.text) + self.assertEqual(pending.json()[0]["task_id"], task_id) + + rejected = self.client.post( + "/api/v1/node/task/report", + json={"sn": "NODE-T", "task_id": task_id, "success": True, "nonce": "0" * 32}, + ) + self.assertEqual(rejected.status_code, 400, rejected.text) + + accepted = self.client.post( + "/api/v1/node/task/report", + json={"sn": "NODE-T", "task_id": task_id, "success": True, "nonce": nonce}, + ) + self.assertEqual(accepted.status_code, 200, accepted.text) + self.assertEqual(accepted.json()["status"], "completed") + + duplicated = self.client.post( + "/api/v1/node/task/report", + json={"sn": "NODE-T", "task_id": task_id, "success": True, "nonce": nonce}, + ) + self.assertEqual(duplicated.status_code, 409, duplicated.text) + + def test_daily_yield_uses_uptime_and_task_response(self) -> None: + yesterday = (utcnow() - timedelta(days=1)).date() + day_start = utcnow().replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=1) + day_end = day_start + timedelta(days=1) + for wallet, sn in (("0xaaa", "NODE-FULL"), ("0xbbb", "NODE-NOTASK"), ("0xccc", "NODE-MISS")): + self._register(wallet, sn, "365") + self._cover_full_day(sn, day_start, day_end) + + with SessionLocal() as session: + full = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-FULL")) + miss = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-MISS")) + session.add(self._task(full.id, TaskStatus.COMPLETED, day_start)) + session.add(self._task(miss.id, TaskStatus.FAILED, day_start)) + session.commit() + + settled = self.client.post("/api/v1/yield/settle", params={"yield_date": yesterday.isoformat()}) + self.assertEqual(settled.status_code, 200, settled.text) + self.assertEqual(settled.json()["created"], 3) + + full_row = self._yield_row("NODE-FULL") + idle_row = self._yield_row("NODE-NOTASK") + miss_row = self._yield_row("NODE-MISS") + self.assertEqual(full_row.online_rate, Decimal("1.00000000")) + self.assertEqual(full_row.task_response_rate, Decimal("1.00000000")) + self.assertEqual(full_row.performance, Decimal("1.00000000")) + self.assertEqual(full_row.points_delta, Decimal("1.00000000")) + self.assertEqual(full_row.compute_output, Decimal("365.00000000")) + self.assertEqual(full_row.online_weight, Decimal("0.70000000")) + + self.assertEqual(idle_row.tasks_dispatched, 0) + self.assertEqual(idle_row.performance, Decimal("1.00000000")) + self.assertEqual(idle_row.points_delta, Decimal("1.00000000")) + self.assertEqual(idle_row.online_weight, Decimal("1.00000000")) + + self.assertEqual(miss_row.task_response_rate, Decimal("0.00000000")) + self.assertEqual(miss_row.performance, Decimal("0.70000000")) + self.assertEqual(miss_row.points_delta, Decimal("0.70000000")) + self.assertEqual(miss_row.compute_output, Decimal("255.50000000")) + + again = self.client.post("/api/v1/yield/settle", params={"yield_date": yesterday.isoformat()}) + self.assertEqual(again.json()["created"], 0) + self.assertEqual(again.json()["skipped"], 3) + + future = self.client.post("/api/v1/yield/settle", params={"yield_date": "2999-01-01"}) + self.assertEqual(future.status_code, 400, future.text) + + def test_open_interval_stops_at_last_heartbeat(self) -> None: + yesterday = (utcnow() - timedelta(days=1)).date() + day_start = utcnow().replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=1) + self._register("0xhalf", "NODE-HALF", "365") + noon = day_start + timedelta(hours=12) + with SessionLocal() as session: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == "NODE-HALF")) + node.licensed_at = day_start + node.license_expires_at = day_start + timedelta(days=365) + node.last_heartbeat_at = noon + session.add(UptimeInterval(node_id=node.id, started_at=day_start, ended_at=None)) + session.commit() + + settled = self.client.post("/api/v1/yield/settle", params={"yield_date": yesterday.isoformat()}) + self.assertEqual(settled.status_code, 200, settled.text) + row = self._yield_row("NODE-HALF") + self.assertEqual(row.online_seconds, 12 * 60 * 60) + self.assertEqual(row.online_rate, Decimal("0.50000000")) + self.assertEqual(row.points_delta, Decimal("0.50000000")) + + def test_daily_cron_writes_previous_closed_day(self) -> None: + # 北京时间 2026-10-12 00:05，对应上一个已结束的 UTC 账本日 2026-10-10。 + moment = datetime(2026, 10, 11, 16, 5, tzinfo=timezone.utc) + day_start = datetime(2026, 10, 10, tzinfo=timezone.utc) + self._register("0xcron", "NODE-CRON", "365") + self._cover_full_day("NODE-CRON", day_start, day_start + timedelta(days=1)) + + summary = run_daily_yield_cron(now=moment) + self.assertEqual(summary.yield_date.isoformat(), "2026-10-10") + self.assertEqual(summary.created, 1) + self.assertFalse(summary.is_partial) + row = self._yield_row("NODE-CRON") + self.assertEqual(row.status, YieldStatus.SETTLED) + self.assertEqual(row.points_delta, Decimal("1.00000000")) + self.assertEqual(row.compute_output, Decimal("365.00000000")) + + again = run_daily_yield_cron(now=moment) + self.assertEqual(again.created, 0) + self.assertEqual(again.skipped, 1) + + def test_withdrawal_reserves_balance_until_rejected_or_confirmed(self) -> None: + yesterday = (utcnow() - timedelta(days=1)).date() + day_start = utcnow().replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=1) + self._register("0xpay", "NODE-PAY", "365") + self._cover_full_day("NODE-PAY", day_start, day_start + timedelta(days=1)) + settled = self.client.post("/api/v1/yield/settle", params={"yield_date": yesterday.isoformat()}) + self.assertEqual(settled.status_code, 200, settled.text) + + too_much = self.client.post( + "/api/v1/withdrawals", + json={"wallet_address": "0xpay", "amount": "10"}, + ) + self.assertEqual(too_much.status_code, 400, too_much.text) + + created = self.client.post( + "/api/v1/withdrawals", + json={"wallet_address": "0xpay", "amount": "0.4", "asset_symbol": "POINT"}, + ) + self.assertEqual(created.status_code, 200, created.text) + withdrawal_id = created.json()["id"] + self.assertEqual(created.json()["available_after"], "0.60000000") + + blocked = self.client.post( + "/api/v1/withdrawals", + json={"wallet_address": "0xpay", "amount": "0.7"}, + ) + self.assertEqual(blocked.status_code, 400, blocked.text) + + confirmed = self.client.post( + f"/api/v1/withdrawals/{withdrawal_id}/confirm", + json={"tx_hash": "0xabc123"}, + ) + self.assertEqual(confirmed.status_code, 200, confirmed.text) + self.assertEqual(confirmed.json()["status"], "completed") + self.assertEqual(confirmed.json()["tx_hash"], "0xabc123") + + balance = self.client.get("/api/v1/users/0xpay/balance") + self.assertEqual(balance.json()["available_points"], "0.60000000") + + def _register(self, wallet: str, sn: str, amount: str) -> None: + user = self.client.post("/api/v1/users", json={"wallet_address": wallet}) + self.assertEqual(user.status_code, 200, user.text) + node = self.client.post( + "/api/v1/nodes", + json={"wallet_address": wallet, "sn": sn, "purchase_amount": amount}, + ) + self.assertEqual(node.status_code, 200, node.text) + self.assertEqual(node.json()["status"], "offline") + self.assertEqual(node.json()["license_months"], 12) + + def _shift_presence(self, sn: str, moment) -> None: + with SessionLocal() as session: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == sn)) + node.last_heartbeat_at = moment + interval = session.scalar( + select(UptimeInterval).where( + UptimeInterval.node_id == node.id, + UptimeInterval.ended_at.is_(None), + ) + ) + interval.started_at = moment + session.commit() + + def _cover_full_day(self, sn: str, day_start, day_end) -> None: + with SessionLocal() as session: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == sn)) + node.licensed_at = day_start + node.license_expires_at = day_start + timedelta(days=365) + session.add(UptimeInterval(node_id=node.id, started_at=day_start, ended_at=day_end)) + session.commit() + + def _task(self, node_id: int, status: TaskStatus, day_start) -> ComputeTask: + return ComputeTask( + id=str(uuid.uuid4()), + node_id=node_id, + task_type="simulated_compute", + payload={"op": "simulated_compute", "nonce": "abc123abc123abcd", "seed": "seedseed"}, + status=status, + push_status=PushStatus.QUEUED, + dispatched_at=day_start + timedelta(hours=3), + deadline_at=day_start + timedelta(hours=4), + success=status == TaskStatus.COMPLETED, + ) + + def _yield_row(self, sn: str) -> NodeYield: + with SessionLocal() as session: + node = session.scalar(select(ComputeNode).where(ComputeNode.sn == sn)) + return session.scalar(select(NodeYield).where(NodeYield.node_id == node.id)) + + +if __name__ == "__main__": + unittest.main() diff --git a/说明.md b/说明.md new file mode 100644 index 0000000..5bce56e --- /dev/null +++ b/说明.md @@ -0,0 +1,110 @@ +# HyperNode-AI 说明 + +HyperNode-AI 是算力节点的控制面。节点程序部署在各处，本服务负责登记节点、接收心跳、下发模拟任务、按天计算积分，并记录提现。 + +时间全部使用 UTC。每日清算的触发时刻是北京时间 00:05，入账的是上一个已经结束的 UTC 日。 + +本地开发直接启动时监听 8000。服务器按 [部署.md](部署.md) 安装后监听 33333。 + +```bash +pip install -r requirements.txt +python -m app +``` + +## 目录 + +| 路径 | 内容 | +|---|---| +| `app/models.py` | 数据表 | +| `app/api.py` | HTTP 接口 | +| `app/services/heartbeat.py` | 心跳与离线判断 | +| `app/services/tasks.py` | 任务下发与 nonce 校验 | +| `app/services/yield_engine.py` | 产出公式与按日入账 | +| `app/jobs/daily_yield.py` | 凌晨清算入口 | +| `app/scheduler.py` | 进程内定时任务 | +| `app/settle.py` | 可单独执行的清算命令 | +| `deploy/ubuntu22.sh` | Ubuntu 22.04 交互式部署 | + +## 数据表 + +业务表： + +- `users`：钱包地址、注册时间。 +- `compute_nodes`：节点 SN、购买金额、12 个月授权、`online` / `offline`、最后心跳。可选 `callback_url` 用来推送任务。 +- `node_yields`：每日算力产出和积分。同一节点同一天只有一条。 +- `withdrawal_logs`：提现金额、资产符号、链上哈希、状态。 + +计算用的表： + +- `uptime_intervals`：一段连续在线时间。 +- `compute_tasks`：下发的模拟任务和回传结果。 +- `scheduler_locks`：多台服务实例同时跑定时任务时互斥。 + +## 产出公式 + +授权金额按授权起止日期的天数线性释放，再乘当天表现系数。单节点累计积分不超过购买金额。 + +- 在线率 = 当天授权窗口内的在线秒数 / 当天授权有效秒数 +- 任务响应率 = 校验通过且执行成功的任务数 / 当天下发任务数 +- 当天没有下发任务时，表现系数 = 在线率 +- 有任务时，表现系数 = `0.7 × 在线率 + 0.3 × 响应率` +- 算力产出 = 购买金额 × 表现系数 +- 当日积分 = 算力产出 / 授权天数 + +开放的在线区间如果最后心跳已超过 15 分钟，只计算到最后一次心跳。 + +权重可用环境变量 `ONLINE_WEIGHT`、`TASK_WEIGHT` 修改，默认 `0.7` 和 `0.3`。 + +## 运行中的三件事 + +服务进程里有三个定时任务： + +1. 每 5 分钟检查心跳。超过 15 分钟没有心跳，或授权到期，节点改为 `offline`，在线区间结束在最后一次心跳。 +2. 每 5 分钟向在线、授权仍有效、且没有未完成任务的节点随机下发任务，每轮最多 20 台。任务带 `task_id` 和一次性 `nonce`。节点若配置了 `callback_url`，会额外 POST 一次；否则节点用拉取接口领取。 +3. 北京时间 00:05 结算上一个已结束的 UTC 日，写入 `node_yields`。同一天已有记录则跳过。 + +回传必须带上任务里的 `nonce`。超时、重复回传、nonce 不匹配都不算成功。 + +## 接口 + +前缀都是 `/api/v1`。金额在 JSON 里是字符串。 + +| 方法 | 路径 | 作用 | +|---|---|---| +| POST | `/users` | 注册钱包 | +| POST | `/nodes` | 登记节点 | +| GET | `/nodes/{sn}` | 查看节点 | +| POST | `/node/heartbeat` | 心跳，请求体 `{"sn":"..."}` | +| GET | `/node/{sn}/tasks` | 拉取未完成任务 | +| POST | `/node/task/report` | 回传 `sn`、`task_id`、`success`、`nonce` | +| GET | `/nodes/{sn}/contribution?yield_date=YYYY-MM-DD` | 预览某日产出，不入账 | +| POST | `/yield/settle?yield_date=YYYY-MM-DD` | 手动清算，不传日期则清算昨天 | +| GET | `/nodes/{sn}/yields` | 节点流水 | +| GET | `/users/{wallet}/balance` | 可用积分 | +| POST | `/withdrawals` | 申请提现 | +| POST | `/withdrawals/{id}/confirm` | 写入 `tx_hash` 并完成 | +| POST | `/withdrawals/{id}/reject` | 拒绝并释放占用的积分 | + +可用积分 = 已入账积分 − 待处理提现 − 已完成提现。 + +健康检查是 `/health`，不在 `/api/v1` 下面。 + +## 配置 + +生产环境的配置在 `/opt/hypernode/.env`。也可以在启动前用同名环境变量覆盖。 + +| 变量 | 默认值 | 含义 | +|---|---|---| +| `DATABASE_URL` | 部署后指向 `/var/lib/hypernode/hypernode.db` | 数据库。可改为 PostgreSQL 连接串 | +| `HEARTBEAT_TIMEOUT_MINUTES` | `15` | 心跳超时 | +| `OFFLINE_CHECK_MINUTES` | `5` | 离线巡检间隔 | +| `TASK_DISPATCH_MINUTES` | `5` | 任务下发间隔 | +| `DISPATCH_BATCH_SIZE` | `20` | 每轮最多下发数量 | +| `TASK_DEADLINE_MINUTES` | `10` | 任务回传时限 | +| `ONLINE_WEIGHT` / `TASK_WEIGHT` | `0.7` / `0.3` | 有任务时的表现系数权重 | +| `YIELD_TIMEZONE` | `Asia/Shanghai` | 清算触发时区 | +| `YIELD_HOUR` / `YIELD_MINUTE` | `0` / `5` | 清算时刻 | +| `LICENSE_MONTHS` | `12` | 新节点授权月数 | +| `DISABLE_SCHEDULER` | `false` | 设为 `true` 时不启动定时任务 | + +修改 `.env` 后执行 `systemctl restart hypernode`。 diff --git a/部署.md b/部署.md new file mode 100644 index 0000000..a27bc95 --- /dev/null +++ b/部署.md @@ -0,0 +1,92 @@ +# HyperNode-AI 部署说明 + +适用于 Ubuntu 22.04。脚本只安装应用本身，监听 `0.0.0.0:33333`。Nginx、Caddy 或其他反代请自行配置，上游地址使用 `127.0.0.1:33333`。 + +## 1. 把代码放到服务器 + +仓库地址：[https://git.bz121.com/dekun/HyperNode-AI.git](https://git.bz121.com/dekun/HyperNode-AI.git) + +```bash +git clone https://git.bz121.com/dekun/HyperNode-AI.git +cd HyperNode-AI +sudo bash deploy/ubuntu22.sh +``` + +私有仓库克隆时需要你的 Gitea 账号。也可以用 scp 把整个项目目录上传到服务器，再进入该目录执行同一条命令。 + +## 2. 菜单 + +脚本是交互式的，启动后只有四个选项： + +| 选项 | 作用 | +|---|---| +| 1 一键部署 | 安装 Python 依赖、创建系统用户、写入 systemd，并启动服务 | +| 2 一键卸载 | 停止服务并删除程序。数据库要再输入一次 `yes` 才会删除 | +| 3 更新 | 用当前目录的代码覆盖 `/opt/hypernode`，重装依赖并重启。`.env` 和数据库保留 | +| 4 退出 | 离开脚本 | + +部署和更新结束时会请求 `http://127.0.0.1:33333/health`，失败会打印最近 50 行服务日志。 + +重复执行「一键部署」不会覆盖已经存在的 `/opt/hypernode/.env`，也不会删除数据库。 + +## 3. 安装结果 + +| 项目 | 位置 | +|---|---| +| 程序 | `/opt/hypernode` | +| 配置 | `/opt/hypernode/.env` | +| 虚拟环境 | `/opt/hypernode/.venv` | +| 数据库 | `/var/lib/hypernode/hypernode.db` | +| systemd | `hypernode.service` | +| 监听 | `0.0.0.0:33333`，进程数固定为 1 | + +进程数必须是 1。每日清算跑在这个进程里，开多个 worker 会重复触发定时任务。 + +常用命令： + +```bash +systemctl status hypernode +journalctl -u hypernode.service -f +curl http://127.0.0.1:33333/health +``` + +接口文档：`http://<服务器 IP>:33333/docs` + +手动补跑某一天的清算： + +```bash +sudo -u hypernode /opt/hypernode/.venv/bin/python -m app.settle --date 2026-10-09 +``` + +不带日期时，结算上一个已经结束的 UTC 账本日。服务正常运行时，北京时间每天 00:05 会自动执行。 + +## 4. 反代 + +脚本不修改 Nginx。同一台机器上的反代把流量转到本机 33333 即可，例如： + +```nginx +location / { + proxy_pass http://127.0.0.1:33333; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; +} +``` + +当前接口没有登录鉴权。反代若只在本机使用，云安全组不要对公网放行 33333。需要从外网直接访问时，再放行该端口，并在反代或安全组上限制来源。 + +## 5. 更新代码 + +在服务器上进入新的项目目录后重新打开菜单，选择 `3`： + +```bash +cd /path/to/HyperNode-AI +sudo bash deploy/ubuntu22.sh +``` + +选择更新后，脚本把该目录同步到 `/opt/hypernode`，然后重启服务。 + +## 6. 卸载 + +菜单选择 `2`。先输入 `yes` 确认删除程序。是否删除 `/var/lib/hypernode` 会再问一次；直接回车会留下数据库。