From db1264953a5548f7cd4e3b17a078ed8091d00d79 Mon Sep 17 00:00:00 2001 From: 12451 <12451@example.com> Date: Sun, 20 Sep 2026 01:21:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(ops):=20=E6=81=A2=E5=A4=8D=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=E4=B8=8E=E8=BF=90=E7=BB=B4=E7=BC=BA=E5=8F=A3=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=EF=BC=88=E5=BB=BA=E8=A1=A8=E9=A1=BA=E5=BA=8F/?= =?UTF-8?q?=E5=A4=87=E4=BB=BD/=E5=BB=BA=E5=8D=95=E6=BC=8F=E5=8D=95/214?= =?UTF-8?q?=E4=BB=93=E4=BD=93/=E9=98=9F=E5=88=97=E4=B8=8A=E9=99=90/NOFILE?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 恢复能力（此前实际为 0） - migrate.rs 的 SCHEMA_SQL 存在前向引用：charge_orders 建表引用其后才创建的 compartments → 全新空库必然报 MySQL 1824，而 main.rs 是 .expect("数据库迁移失败") + Restart=always → 无限崩溃循环。修正为把 cabin_boards/compartments 移到 cabinets 之后 （只挪位置、21 条语句字符数完全一致；实测旧版 20/21 → 新版 21/21， 并用真实二进制在全新空库跑通、建出 21 表 153 列） - 新增每日备份 deploy/backup-db.sh + deploy/pms-backup.cron： pms 与 pms_dev 各一份，--single-transaction --routines --triggers，单库 dump（无 USE/CREATE DATABASE， 演练可安全恢复到临时库），.part 原子 mv、gzip -t 校验、记录表数与精确行数、保留 7 天、flock 防重叠、 任一库失败则跳过清理并 exit 1（避免失败时误删旧备份）。cron 已真实触发验证； 恢复演练：21 表逐表行数一致（1339=1339）后清理无残留 - migrate.rs 每次启动无条件 ALTER cabinets.abstract_id 改为先查 information_schema 再决定， 实测老库触发 1 次、后续启动 0 次 ## 订单与占用 - h5.rs start_charge：校验顺序改为 权限→归属→设备启用→设备在线→建单→下发（原为先建单后校验， 设备离线时会留下幽灵单）；取不到 bat_in 登记（48h TTL 过期）时不再跳过建单，改为按未知电池建单 （与设备侧 handle_device_on 兜底一致）—— 此前会导致整段会话无订单/无记录/无电量 - report_worker.rs 新增无订单却标占用的兜底复位（优先按通道 batt_status bit7，拿不到状态时按 超过 12 小时未更新），并已幂等清零生产历史遗留下的 214 行；两库复查均为 0 ## 队列与资源上限 - device-server：device:reports 加 LPUSH→LLEN→LTRIM 0 199999→LLEN 的 pipeline 与越界 WARN （此前无上限：68 柜停 8 天、1000 柜停 13 小时即打满 1GB Redis，且会连累同实例既有生产平台） - 死信队列加 LTRIM 0 9999 + EXPIRE 7 天，写入时 WARN 带累计死信数 - 两服务各加 600s 一次的队列深度 INFO 日志（可观测性） - 4 个 systemd 单元（仓库 + 线上已安装）加 LimitNOFILE=65535（原软限制 1024，约 1000 柜即 accept 失败） ## 文档 - docs/API.md：接口总表 71 → 79 条并与 mod.rs 的 79 次 .route( 注册双向比对一致； 修正 system:log（种子里不存在的权限码）→ operation_log:view、登出 GET→POST、能耗导出 → energy:export； 补齐 21 个未记录接口；新增"列表响应包装"与"关键数值字段单位"两节 ## 超长函数 - report_worker.rs handle_bat_out 100→63、ensure_charge_order 88→68；users.rs update_user 81→45 （纯重构：字符串字面量多重集零差异，allow/unwrap/expect/panic 计数不变） ## 验收 - cargo check 两服务 0 error/0 warning；cargo test api 90 + device 22 全过 - 前端真实类型检查 npm run typecheck（= tsc -b；原 `tsc --noEmit` 因 tsconfig 是 solution 文件而空跑=假绿） - 端到端联调（测试环境）13/13；生产已部署并验证：服务 active、health 200、ERROR 0、 消费链路正常、214 仓体清零、唯一索引在位 --- deploy/backup-db.sh | 178 ++++++++ deploy/pms-api-prod.service | 2 + deploy/pms-api-test.service | 2 + deploy/pms-backup.cron | 18 + deploy/pms-device-prod.service | 2 + deploy/pms-device-test.service | 2 + docs/API.md | 362 +++++++++++++--- software/api-server/src/db/migrate.rs | 183 +++++++- software/api-server/src/routes/h5.rs | 131 +++--- software/api-server/src/routes/users.rs | 83 +++- .../api-server/src/workers/report_worker.rs | 404 +++++++++++++++--- software/device-server/src/main.rs | 108 ++++- software/device-server/src/tcp/handler.rs | 65 ++- 13 files changed, 1299 insertions(+), 241 deletions(-) create mode 100644 deploy/backup-db.sh create mode 100644 deploy/pms-backup.cron diff --git a/deploy/backup-db.sh b/deploy/backup-db.sh new file mode 100644 index 0000000..180d7b8 --- /dev/null +++ b/deploy/backup-db.sh @@ -0,0 +1,178 @@ +#!/usr/bin/env bash +# +# PMS 数据库每日备份脚本（仓库源文件；部署位置 /usr/local/bin/pms-backup-db.sh） +# +# 背景（2026-09 运维审查）：服务器上此前**没有任何数据库备份作业**， +# `crontab -l` 只有 acme.sh，/etc/cron.daily|weekly 为空，全盘无 mysqldump/xtrabackup 痕迹， +# 恢复能力实际为 0（完全依赖阿里云 RDS 自动备份，且该策略无法从 SQL 读取、无仓库记载）。 +# 本脚本 + /etc/cron.d/pms-backup 补上这一环。 +# +# 设计要点： +# 1. 备份 pms（生产）与 pms_dev（测试）两个库；--single-transaction 一致性快照，不锁表； +# 2. --routines --events --triggers 显式带上：本项目当前只有生成列（mysqldump 自动处理）， +# 无存储过程/触发器/事件，但将来一旦引入必须能一并恢复，故现在就带上防漏； +# 3. --default-character-set=utf8mb4 与库字符集一致，避免中文/emoji 变问号； +# 4. 单库 dump（**不带 --databases**）：dump 内不含 CREATE DATABASE / USE， +# 因此可以安全地恢复到任意临时库做演练（`zcat x.sql.gz | mysql pms_restore_check`）， +# 绝不会因为 dump 里的 USE `pms` 而误写生产库； +# 5. --set-gtid-purged=OFF：RDS 开了 GTID 时，dump 会带 SET @@GLOBAL.GTID_PURGED， +# 恢复到非空实例会直接报 1840 失败；关掉它保证 dump 可移植; +# 6. 密码**不落脚本、不落 cron 行**：统一走 /root/.my.cnf（权限 600，--defaults-extra-file）； +# 7. 落盘 /data/backup/mysql//_<日期>.sql.gz， +# 每个文件记录大小与精确行数（SUM(COUNT(*))）到日志，并做 gzip -t 完整性校验； +# 8. 保留最近 7 天：find ... -mtime +7 -delete（先删过期文件，再删变空的日期目录）； +# 9. 只有**全部库都备份成功**才执行保留策略清理，避免"新备份失败 + 删掉旧备份"= 无备份； +# 10. flock 防止 cron 与手工执行重叠。 +# +# 用法： +# /usr/local/bin/pms-backup-db.sh # 正常备份（cron 每日 03:30 调用） +# tail -n 100 /var/log/pms-backup-db.log # 查看备份日志 +# +set -euo pipefail + +# ---------- 可调参数 ---------- +BACKUP_ROOT="/data/backup/mysql" +RETENTION_DAYS=7 +DATABASES=("pms" "pms_dev") +MYSQL_HOST="127.0.0.1" +MYSQL_PORT="3306" +DEFAULTS_FILE="/root/.my.cnf" +LOG_FILE="/var/log/pms-backup-db.log" +LOCK_FILE="/var/lock/pms-backup-db.lock" + +# ⚠️ --defaults-extra-file 必须是 mysqldump/mysql 的**第一个**参数，否则客户端报错退出 +MYSQL_OPTS=(--defaults-extra-file="$DEFAULTS_FILE" -h "$MYSQL_HOST" -P "$MYSQL_PORT") +DUMP_OPTS=( + --single-transaction + --routines + --events + --triggers + --default-character-set=utf8mb4 + --no-tablespaces + --set-gtid-purged=OFF +) + +DATE_DIR="$(date +%Y-%m-%d)" +DEST_DIR="$BACKUP_ROOT/$DATE_DIR" + +# ---------- 日志 ---------- +# 同时写日志文件与标准输出；cron 行把标准输出丢到 /dev/null，避免日志重复 +log() { + local line + line="$(date '+%Y-%m-%d %H:%M:%S') $*" + printf '%s\n' "$line" >> "$LOG_FILE" + printf '%s\n' "$line" +} + +# ---------- 辅助查询 ---------- +# 统计某库的基础表数量（失败返回 unknown，不中断备份） +count_tables() { + local db="$1" out + if ! out="$(mysql "${MYSQL_OPTS[@]}" -N -B -e \ + "SELECT COUNT(*) FROM information_schema.tables \ + WHERE table_schema='${db}' AND table_type='BASE TABLE'" 2>&1)"; then + log "[warn] 查询 ${db} 表数量失败: ${out}" + echo "unknown" + return 0 + fi + echo "${out:-0}" +} + +# 统计某库所有基础表的**精确**总行数（information_schema.TABLE_ROWS 是估算值，不能用） +# +# ⚠️ mysql 客户端**没有指定默认库**（凭据文件里也不给库），因此每张表必须写成 +# `\`库名\`.\`表名\``；只写 `\`表名\`` 会报 1046 No database selected —— 曾经的实现 +# 把 stderr 丢进 /dev/null，结果每库都静默统计成 0 行。 +exact_row_count() { + local db="$1" tables union_sql rows + if ! tables="$(mysql "${MYSQL_OPTS[@]}" -N -B -e \ + "SELECT table_name FROM information_schema.tables \ + WHERE table_schema='${db}' AND table_type='BASE TABLE'" 2>&1)"; then + log "[warn] 查询 ${db} 表清单失败: ${tables}" + echo "unknown" + return 0 + fi + if [ -z "$tables" ]; then + echo 0 + return 0 + fi + union_sql="" + while IFS= read -r t; do + [ -n "$t" ] && union_sql+="SELECT COUNT(*) AS c FROM \`${db}\`.\`${t}\` UNION ALL " + done <<< "$tables" + if ! rows="$(mysql "${MYSQL_OPTS[@]}" -N -B -e \ + "SELECT SUM(c) FROM (${union_sql%UNION ALL }) AS t" 2>&1)"; then + log "[warn] 统计 ${db} 行数失败: ${rows}" + echo "unknown" + return 0 + fi + echo "${rows:-0}" +} + +# ---------- 单库备份 ---------- +# 备份 -> gzip 中间文件 -> 原子改名 -> 完整性校验 -> 记录大小/表数/行数 +dump_database() { + local db="$1" out="$2" bytes size tables rows + if ! mysqldump "${MYSQL_OPTS[@]}" "${DUMP_OPTS[@]}" "$db" | gzip > "${out}.part"; then + log "[ERROR] ${db} 备份失败（mysqldump/gzip 非 0 退出）" + rm -f "${out}.part" + return 1 + fi + mv -f "${out}.part" "$out" + + if ! gzip -t "$out"; then + log "[ERROR] ${db} 备份 gzip 完整性校验失败: ${out}" + return 1 + fi + + bytes="$(stat -c %s "$out")" + size="$(du -h "$out" | cut -f1)" + tables="$(count_tables "$db")" + rows="$(exact_row_count "$db")" + log "[OK] ${db} -> ${out} 大小=${size}(${bytes} 字节) 表数=${tables} 总行数=${rows} gzip -t=OK" + return 0 +} + +# ---------- 保留策略：只留最近 RETENTION_DAYS 天 ---------- +apply_retention() { + local removed kept + removed="$(find "$BACKUP_ROOT" -mindepth 2 -maxdepth 2 -type f -name '*.sql.gz' \ + -mtime +"${RETENTION_DAYS}" -print -delete | wc -l)" + # 删掉因文件被清理而变空的日期目录 + find "$BACKUP_ROOT" -mindepth 1 -maxdepth 1 -type d -empty -print -delete >> "$LOG_FILE" + kept="$(find "$BACKUP_ROOT" -mindepth 1 -maxdepth 1 -type d | sort | tr '\n' ' ')" + log "[retention] 已删除 ${RETENTION_DAYS} 天前的备份文件 ${removed} 个；当前保留日期目录: ${kept}" +} + +# ---------- 主流程 ---------- +main() { + exec 9>"$LOCK_FILE" + if ! flock -n 9; then + log "[skip] 已有备份任务在运行，本次跳过" + exit 0 + fi + + if [ ! -r "$DEFAULTS_FILE" ]; then + log "[FATAL] 凭据文件 ${DEFAULTS_FILE} 不存在或不可读（应为 600 权限）" + exit 1 + fi + + mkdir -p "$DEST_DIR" + log "[start] 备份开始，目标目录 ${DEST_DIR}，库: ${DATABASES[*]}" + + local failed=0 db + for db in "${DATABASES[@]}"; do + dump_database "$db" "$DEST_DIR/${db}_${DATE_DIR}.sql.gz" || failed=1 + done + + if [ "$failed" -eq 0 ]; then + apply_retention + log "[done] 全部备份成功" + else + # 关键顺序：备份失败时**不能**执行清理，否则可能出现"新备份没有、旧备份被删" + log "[done] 存在失败的备份，已跳过保留策略清理（避免误删仅存可用备份）" + fi + exit "$failed" +} + +main "$@" diff --git a/deploy/pms-api-prod.service b/deploy/pms-api-prod.service index b9842b6..03ce55f 100644 --- a/deploy/pms-api-prod.service +++ b/deploy/pms-api-prod.service @@ -10,6 +10,8 @@ ExecStart=/data/pms/api-server/pms-api EnvironmentFile=/data/pms/api-server/.env Restart=always RestartSec=5 +# 提高文件描述符上限：默认软限制 1024 在大规模设备接入时会先耗尽 +LimitNOFILE=65535 [Install] WantedBy=multi-user.target diff --git a/deploy/pms-api-test.service b/deploy/pms-api-test.service index 118096f..45de2f7 100644 --- a/deploy/pms-api-test.service +++ b/deploy/pms-api-test.service @@ -10,6 +10,8 @@ ExecStart=/data/pms_dev/api-server/pms-api EnvironmentFile=/data/pms_dev/api-server/.env Restart=always RestartSec=5 +# 提高文件描述符上限：默认软限制 1024 在大规模设备接入时会先耗尽 +LimitNOFILE=65535 [Install] WantedBy=multi-user.target diff --git a/deploy/pms-backup.cron b/deploy/pms-backup.cron new file mode 100644 index 0000000..ea40a43 --- /dev/null +++ b/deploy/pms-backup.cron @@ -0,0 +1,18 @@ +# PMS 数据库每日备份 — cron 定义（仓库源文件；部署位置 /etc/cron.d/pms-backup） +# +# 安装： +# install -m 0644 deploy/pms-backup.cron /etc/cron.d/pms-backup +# install -m 0755 deploy/backup-db.sh /usr/local/bin/pms-backup-db.sh +# 验证：cat /etc/cron.d/pms-backup && /usr/local/bin/pms-backup-db.sh +# +# 说明： +# * /etc/cron.d/ 的文件必须带「用户」字段（这里的 root），且不能 group/other 可写； +# * 密码不在本行、也不在脚本里 —— 走 /root/.my.cnf（600）； +# * 脚本自身把明细写 /var/log/pms-backup-db.log；cron 这里只兜住脚本意外崩溃的 stderr， +# 避免与脚本日志重复。 +SHELL=/bin/bash +PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin +MAILTO="" + +# 每日 03:30 备份 pms(生产) + pms_dev(测试)，保留最近 7 天 +30 3 * * * root /usr/local/bin/pms-backup-db.sh >/dev/null 2>>/var/log/pms-backup-db.cron.log diff --git a/deploy/pms-device-prod.service b/deploy/pms-device-prod.service index e4d9e52..db9eaf6 100644 --- a/deploy/pms-device-prod.service +++ b/deploy/pms-device-prod.service @@ -10,6 +10,8 @@ ExecStart=/data/pms/device-server/pms-device-server EnvironmentFile=/data/pms/device-server/.env Restart=always RestartSec=5 +# 提高文件描述符上限：默认软限制 1024 ⇒ 约 1000 台柜子就 accept() 失败 +LimitNOFILE=65535 [Install] WantedBy=multi-user.target diff --git a/deploy/pms-device-test.service b/deploy/pms-device-test.service index 10ea948..aaae25d 100644 --- a/deploy/pms-device-test.service +++ b/deploy/pms-device-test.service @@ -10,6 +10,8 @@ ExecStart=/data/pms_dev/device-server/device-server EnvironmentFile=/data/pms_dev/device-server/.env Restart=always RestartSec=5 +# 提高文件描述符上限：默认软限制 1024 ⇒ 约 1000 台柜子就 accept() 失败 +LimitNOFILE=65535 [Install] WantedBy=multi-user.target diff --git a/docs/API.md b/docs/API.md index 1f1ef9a..83bf369 100644 --- a/docs/API.md +++ b/docs/API.md @@ -11,79 +11,140 @@ ## 统一端点清单 +> **唯一事实来源**：`software/api-server/src/routes/mod.rs` 的 `.route(...)` 注册清单。 +> 本表**一行 = 一次 `.route(...)` 注册**，共 **79** 行，与 +> `Select-String -Path software\api-server\src\routes\mod.rs -Pattern '\.route\(' | Measure-Object` +> 得到的 **79** 条完全一致。同一次注册里链式绑定多个方法时，"方法"列列出全部方法（如 `GET / POST`）， +> 权限列按方法分别标注。 +> +> 权限码**一律取自 `db/migrate.rs::seed_permissions` 的种子表**（共 31 个权限码）。 +> 处理器内部没有调用 `check_permission` 的接口，如实标注为**"仅需登录"**（只过认证中间件） +> 或**"免认证"**（公开路由），不臆造权限码。 +> +> ⚠️ `mod.rs` 里还有一条 `.nest_service("/uploads", ServeDir::new("uploads"))`（静态头像目录，**免认证**）， +> 它不是 `.route(...)` 注册，故不计入上表的 79 条。 + | 方法 | 路径 | 说明 | 权限 | |------|------|------|------| -| POST | `/api/auth/login` | 登录 | 免认证 | -| GET | `/api/auth/logout` | 登出 | 需认证 | -| GET | `/api/auth/me` | 当前用户信息 | 需认证 | -| GET | `/api/dashboard` | 仪表盘数据 | 需认证 | -| GET | `/api/cabinets` | 设备列表（含 RSSI） | `device:view` | +| POST | `/api/auth/logout` | 登出 | 仅需登录 | +| GET | `/api/auth/me` | 当前用户信息 | 仅需登录 | +| PUT | `/api/auth/avatar` | 更新头像 URL | 仅需登录 | +| POST | `/api/auth/avatar/upload` | 上传头像文件（multipart） | 仅需登录 | +| POST | `/api/devices/{dev_id}/command` | 下发设备指令 | `device:operate` | +| GET | `/api/devices/stats` | 设备服务节点统计 | `device:view` | +| POST | `/api/nodes/register` | 设备服务节点注册（device-server 调用） | 仅需登录 | +| POST | `/api/nodes/{node_id}/heartbeat` | 节点心跳 | 仅需登录 | +| POST | `/api/nodes/{node_id}/deregister` | 节点注销 | 仅需登录 | +| GET | `/api/nodes` | 节点列表 | 仅需登录 | +| GET / POST | `/api/roles` | 角色列表 / 创建角色 | GET 仅需登录；POST `role:create` | +| PUT / DELETE | `/api/roles/{id}` | 更新 / 删除角色 | `role:edit` / `role:delete` | +| PUT | `/api/roles/{id}/permissions` | 角色赋权 | `role:edit` | +| GET | `/api/permissions` | 权限码列表 | 仅需登录 | +| GET | `/api/users/{id}/permissions` | 用户权限（角色合并结果） | `user:view` | +| PUT | `/api/users/{id}/roles` | 分配角色 | `user:edit` | +| PUT / GET | `/api/users/{id}/scopes` | 设置 / 查询用户数据范围 | PUT `user:edit`；GET `user:view` | +| GET / POST | `/api/organizations` | 组织列表 / 创建组织 | `org:view` / `org:create` | +| PUT / DELETE | `/api/organizations/{id}` | 更新 / 删除组织 | `org:edit` / `org:delete` | +| PUT | `/api/organizations/{id}/status` | 禁用 / 启用组织 | 仅总管理员（代码内 `role_level=2` 判断，**非权限码**） | +| GET / POST | `/api/organizations/{org_id}/projects` | 项目列表 / 创建项目 | `project:view` / `project:create` | +| PUT / DELETE | `/api/projects/{id}` | 更新 / 删除项目 | `project:edit` / `project:delete` | +| GET | `/api/projects/{project_id}/cabinets` | 项目下设备列表 | `device:view` | | POST | `/api/cabinets` | 批量创建柜子 | `device:create` | -| GET | `/api/cabinets/:id` | 设备详情（含仓板+仓体） | `device:view` | -| PUT | `/api/cabinets/:id` | 更新柜子 | `device:edit` | -| DELETE | `/api/cabinets/:id` | 删除柜子 | `device:delete` | -| POST | `/api/cabinets/:id/regenerate-auth` | 重置安全码（仅总管理员） | `device:auth_str:reset` | -| POST | `/api/cabinets/:id/bind` | 绑定设备到项目 | `device:edit` | -| POST | `/api/cabinets/:id/replace-imei` | 更换 IMEI | `device:edit` | -| GET | `/api/cabinets/abstract/:abstract_id` | 扫码查设备（免认证） | 免认证 | -| GET | `/api/cabinets/scan/:abstract_id` | 扫码验证（需认证） | 需认证 | -| GET | `/api/cabinets/filter` | 设备列表（管理端） | 需认证 | -| GET | `/api/cabinets/options` | 设备下拉选项 | 需认证 | -| GET | `/api/cabin-boards/options` | 仓控板下拉选项 | 需认证 | -| GET | `/api/compartments/options` | 仓体下拉选项 | 需认证 | -| GET | `/api/cabinets/realtime-status` | 设备实时状态 | 需认证 | -| GET | `/api/compartments/:id` | 仓体/通道详情 | 需认证 | -| POST | `/api/charge/start` | 开始充电 | `charge:start` | -| POST | `/api/charge/stop` | 停止充电 | `charge:stop` | -| POST | `/api/door/open` | 开门（**柜级**：整柜共用一扇门，只解锁需人拉开） | `charge:open_door` | -| POST | `/api/cabinets/{id}/open` | 解锁柜门（柜级，同上；管理端设备详情页用） | `charge:open_door` | +| PUT / DELETE / GET | `/api/cabinets/{id}` | 更新 / 删除 / 设备详情（含仓板+仓体） | `device:edit` / `device:delete` / `device:view` | +| PUT | `/api/cabinets/{id}/status` | 启用 / 停用柜子 | `device:edit` | +| PUT | `/api/cabinets/{id}/work-mode` | 设置充电工作模式（0=平台指令，1=自动） | `device:edit` | +| POST | `/api/cabinets/{id}/regenerate-auth` | 重置安全码 | `device:auth_str:reset`（仅总管理员） | +| POST | `/api/cabinets/{id}/open` | 解锁柜门（柜级，管理端设备详情页用） | `charge:open_door` | | POST | `/api/cabinets/{id}/pow-on` | 接触器合闸（柜级） | `device:operate` | | POST | `/api/cabinets/{id}/pow-off` | 接触器分闸（柜级） | `device:operate` | | POST | `/api/cabinets/{id}/reset` | 重启整柜（柜级） | `device:operate` | | POST | `/api/cabin-boards/{id}/reset` | 重启单块仓控板 | `device:operate` | -| GET | `/api/organizations` | 组织列表 | `org:view` | -| POST | `/api/organizations` | 创建组织 | `org:create` | -| PUT | `/api/organizations/:id` | 更新组织 | `org:edit` | -| DELETE | `/api/organizations/:id` | 删除组织 | `org:delete` | -| GET | `/api/organizations/:org_id/projects` | 项目列表 | 需认证 | -| POST | `/api/organizations/:org_id/projects` | 创建项目 | `project:create` | -| PUT | `/api/projects/:id` | 更新项目 | `project:edit` | -| DELETE | `/api/projects/:id` | 删除项目 | `project:delete` | -| GET | `/api/organization-tree` | 组织树（含项目/设备） | 需认证 | -| GET | `/api/charge-records` | 充电记录 | `charge_record:view` | -| GET | `/api/charge-records/export` | 导出充电记录 | `charge_record:export` | +| POST | `/api/cabinets/{id}/bind` | 绑定设备到项目 | `device:edit` | +| POST | `/api/cabinets/{id}/replace-imei` | 更换 IMEI | `device:edit` | +| GET | `/api/organization-tree` | 组织树（含项目/设备） | `device:view` | +| GET | `/api/cabinets/filter` | 设备列表（管理端，**响应为 `{total,data}`**，见下） | `device:view` | +| GET | `/api/cabinets/realtime-status` | 单设备实时状态（板级+通道级） | `device:view` | +| GET | `/api/cabinets/options` | 设备下拉选项 | `device:view` | +| GET | `/api/cabin-boards/options` | 仓控板下拉选项 | `device:view` | +| GET | `/api/compartments/options` | 仓体下拉选项 | `device:view` | +| GET | `/api/charge-records` | 充电记录列表 | `charge_record:view` | +| GET | `/api/charge-records/export` | 导出充电记录（异步任务） | `charge_record:export` | | GET | `/api/charge-orders` | 充电订单列表 | `charge_record:view` | | GET | `/api/batteries` | 电池台账列表 | `device:view` | +| GET | `/api/device-logs` | 设备日志 | `device_log:view` | +| GET | `/api/downloads` | 下载中心列表 | `charge_record:export` | +| GET | `/api/downloads/{id}` | 下载任务详情 | `charge_record:export` | +| GET | `/api/downloads/{id}/file` | 下载文件 | `charge_record:export` | +| POST | `/api/downloads/cleanup` | 清理过期下载 | `charge_record:export` | | GET | `/api/energy-stats/summary` | 能耗概览 | `energy:view` | | GET | `/api/energy-stats/cabinets` | 柜子能耗排行 | `energy:view` | | GET | `/api/energy-stats/projects` | 项目能耗统计 | `energy:view` | | GET | `/api/energy-stats/trend` | 能耗趋势 | `energy:view` | -| GET | `/api/energy-stats/export` | 导出能耗数据 | `energy:view` | -| GET | `/api/device-logs` | 设备日志 | `device:view` | -| GET | `/api/operation-logs` | 操作日志 | `system:log` | -| GET | `/api/users` | 用户列表 | `user:view` | -| POST | `/api/users` | 创建用户 | `user:create` | -| PUT | `/api/users/:id` | 更新用户 | `user:edit` | -| DELETE | `/api/users/:id` | 删除用户 | `user:delete` | -| PUT | `/api/users/:id/reset-password` | 重置密码 | `user:edit` | -| PUT | `/api/users/:id/status` | 设置用户状态 | `user:edit` | -| GET | `/api/users/:id/permissions` | 用户权限 | `user:view` | -| PUT | `/api/users/:id/roles` | 分配角色 | `user:edit` | -| GET | `/api/users/:id/scopes` | 用户数据范围 | `user:view` | -| PUT | `/api/users/:id/scopes` | 设置数据范围 | `user:edit` | -| GET | `/api/roles` | 角色列表 | `role:view` | -| POST | `/api/roles` | 创建角色 | `role:create` | -| PUT | `/api/roles/:id` | 更新角色 | `role:edit` | -| DELETE | `/api/roles/:id` | 删除角色 | `role:delete` | -| PUT | `/api/roles/:id/permissions` | 角色赋权 | `role:edit` | -| GET | `/api/permissions` | 权限码列表 | `role:view` | -| GET | `/api/downloads` | 下载中心 | 需认证 | -| GET | `/api/downloads/:id` | 下载详情 | 需认证 | -| GET | `/api/downloads/:id/file` | 下载文件 | 需认证 | -| POST | `/api/downloads/cleanup` | 清理过期下载 | 需认证 | -| POST | `/api/devices/:dev_id/command` | 下发设备指令 | `device:operate` | -| GET | `/api/devices/stats` | 设备服务统计 | 需认证 | +| GET | `/api/energy-stats/export` | 导出能耗数据（异步任务） | **`energy:export`** | +| GET / POST | `/api/users` | 用户列表 / 创建用户 | `user:view` / `user:create` | +| PUT / DELETE | `/api/users/{id}` | 更新 / 删除用户 | `user:edit` / `user:delete` | +| PUT | `/api/users/{id}/reset-password` | 重置密码 | `user:edit` | +| PUT | `/api/users/{id}/status` | 设置用户状态 | `user:edit` | +| GET | `/api/operation-logs` | 操作日志 | **`operation_log:view`** | +| GET | `/api/dashboard` | 小程序首页仪表盘 | 仅需登录（内部按角色做组织隔离） | +| GET | `/api/cabinets` | 设备列表（含实时 RSSI） | `device:view` | +| GET | `/api/compartments/{id}` | 仓体/通道详情（实时 BMS + 充电记录） | `device:view` | +| POST | `/api/charge/start` | 开始充电 | `charge:start` | +| POST | `/api/charge/stop` | 停止充电 | `charge:stop` | +| POST | `/api/door/open` | 开门（**柜级**：整柜共用一扇门，只解锁需人拉开） | `charge:open_door` | +| GET | `/api/cabinets/scan/{abstract_id}` | 扫码验证 | 仅需登录（组织归属校验，**非权限码**） | +| POST / GET | `/api/feedback` | 提交问题反馈 / 反馈列表 | POST 仅需登录；GET `operation_log:view` | +| PUT | `/api/feedback/{id}` | 更新反馈处理状态 | `operation_log:view` | +| POST | `/api/feedback/upload` | 上传反馈图片（multipart） | 仅需登录 | | GET | `/api/health` | 健康检查 | 免认证 | +| GET | `/api/auth/challenge` | 获取登录挑战码 | 免认证 | +| POST | `/api/auth/login` | 登录 | 免认证 | +| POST | `/api/auth/wx-login` | 微信小程序手机号快速登录 | 免认证 | +| GET | `/api/cabinets/abstract/{abstract_id}` | 扫码查设备（二维码入口） | 免认证 | +| GET / POST | `/api/wechat/callback` | 微信服务器校验 / 消息回调 | 免认证 | +| GET | `/api/maps/suggestion` | 腾讯地图输入提示代理 | 免认证 | +| POST | `/api/maps/geocoder` | 腾讯地图逆地址解析代理 | 免认证 | +| GET | `/api/maps/place-search` | 腾讯地图地点搜索代理 | 免认证 | + +### 列表响应包装（与实现逐条核对） + +`data` 字段的结构**并非全站统一**，按实现如实记录如下： + +| 接口 | `data` 结构 | 备注 | +|------|-------------|------| +| `GET /api/charge-records` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/charge-orders` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/batteries` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/device-logs` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/downloads` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/feedback` | `{list,total,page,page_size}` | 标准分页 | +| `GET /api/users` | `{list,total}` | ⚠️ **没有 `page` / `page_size`** | +| `GET /api/operation-logs` | `{list,total}` | ⚠️ **没有 `page` / `page_size`** | +| `GET /api/cabinets/filter` | `{total,data}` | ⚠️ **与其它列表不同**：列表键名是 `data` 而不是 `list`，且无 `page` / `page_size` | +| `GET /api/energy-stats/cabinets` | `[...]` 数组 | ⚠️ `data` 直接是数组，无分页包装 | +| `GET /api/energy-stats/projects` | `[...]` 数组 | ⚠️ `data` 直接是数组，无分页包装 | +| `GET /api/energy-stats/trend` | `{granularity,list}` | ⚠️ 无 `total` / `page` / `page_size` | +| `GET /api/roles`、`GET /api/permissions` | `[...]` 数组 | ⚠️ `data` 直接是数组 | +| `GET /api/organization-tree` | `[...]` 树形数组 | 根节点数组 | + +> ⚠️ **内部接口例外**：`/api/nodes/*` 与 `/api/devices/*` 直接返回 `{"suc":1}` / +> `{"nodes":[...],"count":N}` / `{"node_count":N}`，**不带** `{code,data,message}` 外层 —— 它们面向 +> device-server 内部调用，不是给前端消费的标准响应。 + +### 关键数值字段单位（抽查结论） + +| 字段 | 出现位置 | 单位 | 前端显示 | +|------|----------|------|----------| +| `energy` | 充电记录 / 充电订单 / 能耗统计 | **Wh**（整数） | `/ 1000` → kWh | +| `today_energy` / `month_energy` | 能耗概览 | **Wh**（整数） | `/ 1000` → kWh | +| `max_temperature` | 充电记录 | **K×100**（整数） | `/ 100 - 273.15` → °C | +| `soc` / `start_soc` / `end_soc` | 电池台账 / 充电订单 / 通道状态 | **‰**（整数，0–10000） | `/ 10` → % | +| `voltage` / `cell_voltages[]` | 仓体详情 / BMS | **mV** | `/ 1000` → V | +| `current` | 仓体详情 / BMS | **mA** | `/ 1000` → A | +| `rssi` | 设备列表 / 通道状态 | **0–99 信号等级**（不是 dBm） | 直接显示 | +| `estimated_cost` | 能耗概览 | **元**（`f64`，业务计算值） | 不转换 | + +> 完整单位规范见文末 [数据单位规范](#数据单位规范)；后端**不做任何单位换算**，一律返回原始整数值。 --- @@ -120,6 +181,55 @@ **角色等级：** 0=操作员，1=企业管理员，2=总管理员 +> ⚠️ 登录**必须**带滑块验证码：先调 `GET /api/auth/challenge` 拿 `challenge_id`，拖动对齐后把 +> `challenge_id` + `captcha_x`（偏移量）一并提交给 `POST /api/auth/login`； +> 缺 `challenge_id` 直接 400「缺少登录验证参数」，偏移量校验不过返回 400「验证码错误，请刷新页面重试」。 + +### GET /api/auth/challenge + +获取登录滑块验证码（**免认证**）。返回带缺口的背景图与拼图块图（data URL）。 + +```json +{ + "code": 0, + "data": { + "challenge_id": "b1f0...", + "bg_image": "data:image/png;base64,...", + "piece_image": "data:image/png;base64,..." + }, + "message": "ok" +} +``` + +### POST /api/auth/wx-login + +微信小程序手机号快速登录（**免认证**）。前端先 `wx.login()` 拿 `code`、`getPhoneNumber` 拿 `phone_code`； +后端向微信换取 `openid` + 手机号，按**手机号**匹配用户后签发 JWT，并把 `openid` 写回 `users.openid`。 + +**请求：** `{ "code": "081...", "phone_code": "e3f..." }` + +**响应：** 与 `POST /api/auth/login` 完全同构（`{token, user:{id,name,phone,role_level,organization_id,permissions}}`）， +`message` 为「登录成功」；账户被禁用返回 403「账户已被禁用」。 + +### PUT /api/auth/avatar + +更新当前用户头像 URL（仅需登录）。**请求：** `{ "avatar": "https://.../a.png" }`； +**响应：** `{ "code": 0, "data": { "avatar": "https://.../a.png" }, "message": "ok" }`。 + +### POST /api/auth/avatar/upload + +上传头像文件（仅需登录，`multipart/form-data`）。 + +- 大小上限 **2MB**，超限 400「头像文件不能超过 2MB」；内容为空 400「上传内容为空」； +- 扩展名白名单 `.jpg/.jpeg/.png/.gif/.webp/.svg`，**不在白名单内一律回落为 `.jpg`**（不是报错）； +- 无扩展名时默认 `.jpg`。 + +**响应：** `{ "code": 0, "data": { "avatar": "/uploads/avatars/.jpg" }, "message": "ok" }` +（文件落盘为 `uploads/avatars/`，同时写回 `users.avatar`） + +> ⚠️ 静态文件目录由 `mod.rs` 的 `.nest_service("/uploads", ServeDir::new("uploads"))` 提供，**免认证**。 +> 代码里该处理器的文档注释误写成 `GET`，**实际注册的是 `POST`**（以 `mod.rs` 为准）。 + --- ## 小程序专用接口 @@ -170,7 +280,7 @@ } ``` -### GET /api/cabinets/abstract/:abstract_id +### GET /api/cabinets/abstract/{abstract_id} 扫码入口，免认证。通过二维码中的抽象ID查找设备。 @@ -199,7 +309,7 @@ } ``` -### GET /api/compartments/:id +### GET /api/compartments/{id} 仓体/通道详情，含实时 BMS 数据和充电记录。 @@ -228,9 +338,21 @@ > 权限：`charge:open_door` + 组织隔离 -### GET /api/cabinets/scan/:abstract_id +### GET /api/cabinets/scan/{abstract_id} -扫码验证，需认证。验证当前用户是否有权访问该设备。 +扫码验证，需认证。验证当前用户是否有权访问该设备（校验**组织归属**；无独立权限码）。 + +### 问题反馈 + +| 接口 | 说明 | 权限 | +|------|------|------| +| `POST /api/feedback` | 提交反馈（`content` 必填，≤2000 字，可带 `images`） | 仅需登录（任何已登录用户可提交） | +| `GET /api/feedback` | 反馈列表（`{list,total,page,page_size}`） | `operation_log:view` | +| `PUT /api/feedback/{id}` | 更新处理状态 | `operation_log:view` | +| `POST /api/feedback/upload` | 上传反馈图片（`multipart/form-data`） | 仅需登录 | + +`POST /api/feedback` 响应：`{ "code": 0, "data": { "success": true }, "message": ... }`； +`POST /api/feedback/upload` 响应：`{ "code": 0, "data": { "url": "..." }, "message": "ok" }`。 --- @@ -285,6 +407,52 @@ **状态码：** 0=充电中，1=正常结束，2=异常中断，3=手动停止 +**单位：** `energy` = **Wh**（整数，前端 `/ 1000` 显示 kWh）；`max_temperature` = **K×100**（整数，`/ 100 - 273.15` 显示 °C）。 +`start_time` / `end_time` 为字符串时间，非数值单位。 + +### GET /api/cabinets/filter + +设备列表（管理端）。**响应包装与其它列表接口不同**：`data` 是 `{total,data}`， +列表键名是 `data` 而**不是** `list`，且**没有** `page` / `page_size`。 + +```json +{ + "code": 0, + "data": { + "total": 2, + "data": [ + { "id": 1, "abstract_id": "10-00000000", "name": "太湖新城换电柜A", "status": 1, "online": 1, "rssi": 85 } + ] + }, + "message": "ok" +} +``` + +> ⚠️ 前端解析时不要套用 `{list,total,page,page_size}` 的通用分页逻辑。 + +### PUT /api/cabinets/{id}/work-mode + +设置柜子充电工作模式。权限 `device:edit` + 组织归属校验。 + +**请求：** `{ "work_mode": "0" }`（**字符串**，前端契约）—— 白名单取值，`"0"`=平台指令模式（默认）、 +`"1"`=自动充电模式；其它值（含 `"2"`/`"true"`）返回 400。 + +**响应：** `{ "code": 0, "data": { "success": true }, "message": "ok" }` + +### PUT /api/cabinets/{id}/status + +禁用 / 启用柜子。权限 `device:edit` + 组织归属校验，**且仅总管理员（`role_level=2`）可执行**， +否则 403「仅总管理员可禁用/启用柜子」。 + +**请求：** `{ "status": 2 }`；**响应：** `{ "code": 0, "data": null, "message": "ok" }` + +### PUT /api/organizations/{id}/status + +禁用 / 启用组织。**仅总管理员**（代码内 `role_level < 2` 判断，**不走权限码**）； +禁用后该组织及其子组织的柜子全部不可用（指令链路拦截）。 + +**请求：** `{ "disabled": 1 }`（only `0`/`1`，其它值 400）；**响应：** `{ "code": 0, "data": null, "message": "禁用成功" }` + ### GET /api/charge-records/export 异步导出充电记录 Excel。创建导出任务后自行去下载中心取文件。 @@ -329,8 +497,8 @@ "battery_id": "BAT001", "start_time": "2026-07-05 10:00:00", "end_time": "2026-07-05 12:00:00", - "start_soc": 20, - "end_soc": 95, + "start_soc": 200, + "end_soc": 950, "energy": 1200, "reason": null, "reason_text": "-", @@ -349,6 +517,8 @@ **订单状态：** 0=充电中，1=正常结束，2=手动停止，3=异常终止，4=拔出结束 +**单位：** `energy` = **Wh**（整数）；`start_soc` / `end_soc` = **‰**（整数，0–10000，前端 `/ 10` 显示 %）。 + ### GET /api/batteries 电池台账列表。记录每块电池的插入/拔出事件和当前状态。 @@ -375,7 +545,7 @@ "brand": "星恒", "model": "XH-48V", "voltage": 52000, - "soc": 45, + "soc": 4500, "status": 1, "cabinet_id": 1, "compartment_id": 1, @@ -391,6 +561,9 @@ } ``` +**单位：** `voltage` = **mV**（52000 = 52.000V）；`soc` = **‰**（4500 = 45.00%，表字段注释即 `电量(‰)`）。 +⚠️ 历史文档曾把示例写成 `"soc": 45`，那是按"百分比"写的**错误示例**，实现返回的是千分比。 + ### GET /api/energy-stats/summary 能耗概览，按组织数据隔离返回今日/本月能耗数据。 @@ -416,6 +589,45 @@ --- +## 免认证接口与内部接口 + +> 本节全部来自 `mod.rs::public_routes()`（**免认证**，不经过认证中间件）与节点管理接口。 + +### GET /api/health + +健康检查（免认证）。 + +### GET / POST /api/wechat/callback + +微信服务器回调（免认证）：`GET` 做服务器地址校验（回显 `echostr`），`POST` 接收消息（明文 XML）。 +Nginx 有独立的 `/api/wechat/` location 直连 API 服务，不经 `/api/pms/` 前缀重写。 + +### 腾讯地图代理（免认证，前端需要） + +| 接口 | 方法 | 说明 | +|------|------|------| +| `/api/maps/suggestion` | GET | 输入提示（keyword 联想） | +| `/api/maps/geocoder` | POST | 逆地址解析（经纬度 → 地址） | +| `/api/maps/place-search` | GET | 地点搜索 | + +> 代理的目的：把腾讯地图 Key 留在服务端，前端不暴露 Key。 + +### 设备服务节点与指令接口（响应不含标准外层） + +| 接口 | 方法 | 权限 | 响应 | +|------|------|------|------| +| `/api/nodes/register` | POST | 仅需登录 | `{"suc":1}` | +| `/api/nodes/{node_id}/heartbeat` | POST | 仅需登录 | `{"suc":1}`；未注册 `{"suc":0,"err":"节点未注册"}` | +| `/api/nodes/{node_id}/deregister` | POST | 仅需登录 | `{"suc":1}` | +| `/api/nodes` | GET | 仅需登录 | `{"nodes":[...],"count":N}` | +| `/api/devices/stats` | GET | `device:view` | `{"node_count":N}` | +| `/api/devices/{dev_id}/command` | POST | `device:operate` | `{"suc":1}` | + +> `nodes/*` 是 device-server 内部调用；`devices/*` 是管理端/后端调用。两者都**不返回** +> `{code,data,message}` 标准外层（见上表"内部接口例外"）。 + +--- + ## 数据单位规范 **核心原则：所有数值字段使用整数存储，禁止浮点类型 (f32/f64/DECIMAL/FLOAT)** @@ -755,8 +967,19 @@ device:{imei} # 设备级 Hash ## 权限码列表 -> 以 `software/api-server/src/db/migrate.rs::seed_permissions` 的**种子表为准**（下表与之对齐）。 +> 以 `software/api-server/src/db/migrate.rs::seed_permissions` 的**种子表为准**（下表与之对齐，共 **31** 个权限码）。 > ⚠️ 历史上代码里曾误用一个**不存在**的权限码 `device:door:unlock`（种子里从来没有它）→ 导致开门对所有角色 403，**已修正为 `charge:open_door`**。 +> ⚠️ 同样地，**不存在** `system:log` 这个权限码（文档曾误写）——操作日志用的是 **`operation_log:view`**； +> 照 `system:log` 配角色会导致该角色**永远看不到操作日志**。 +> +> **默认角色授权（`seed_roles`）**： +> - `super_admin`（总管理员）：`grant_all_permissions` —— 种子表全部 31 个； +> - `org_admin`（企业管理员）：24 个 —— `device:view/operate/create/edit`、`charge:start/stop/open_door`、 +> `charge_record:view/export`、`device_log:view`、`energy:view/export`、`user:view/create/edit`、 +> `operation_log:view`、`org:*`(4)、`project:*`(4)； +> **不含** `device:delete`、`device:auth_str:reset`、`user:delete`、`role:*`； +> - `normal_user`（操作员）：8 个 —— `device:view/operate`、`charge:start/stop/open_door`、 +> `charge_record:view`、`device_log:view`、`energy:view`。 | 权限码 | 说明 | |--------|------| @@ -770,11 +993,12 @@ device:{imei} # 设备级 Hash | `charge:stop` | 停止充电 | | `charge:open_door` | 开门（柜门解锁） | | `charge_record:view` | 查看充电记录 | -| `charge_record:export` | 导出充电记录 | +| `charge_record:export` | 导出充电记录（**也用于下载中心 4 个接口**） | | `device_log:view` | 查看设备日志 | | `energy:view` | 查看能耗 | | `energy:export` | 导出能耗 | | `user:view` / `user:create` / `user:edit` / `user:delete` | 用户管理 | | `role:view` / `role:create` / `role:edit` / `role:delete` | 角色管理 | -| `operation_log:view` | 查看操作日志 | +| `operation_log:view` | 查看操作日志（**也用于问题反馈列表/处理状态 2 个接口**） | | `org:view` / `org:create` / `org:edit` / `org:delete` | 组织管理 | +| `project:view` / `project:create` / `project:edit` / `project:delete` | 项目管理 | diff --git a/software/api-server/src/db/migrate.rs b/software/api-server/src/db/migrate.rs index 41f34d5..23bdf84 100644 --- a/software/api-server/src/db/migrate.rs +++ b/software/api-server/src/db/migrate.rs @@ -4,6 +4,14 @@ use sqlx::MySqlPool; /// 全量建表语句（幂等：全部 `CREATE TABLE IF NOT EXISTS`）。 /// 抽成常量是为了让 [`run`] 保持可读（此前 280 行的函数里有 233 行是这段 SQL）。 +/// +/// ⚠️ **语句顺序即依赖顺序**：本常量由 [`run`] 按 `;` 逐条执行，MySQL 的 +/// `FOREIGN KEY ... REFERENCES` 要求被引用表**必须已存在**（否则报 1824 +/// `Failed to open the referenced table`）。因此每张表都必须排在所有引用它的表之前， +/// 顺序为：organizations → projects → cabinets → cabin_boards → compartments → +/// 其余业务表（含引用 compartments 的 charge_orders / charge_records）。 +/// 调整顺序时只允许整块移动 CREATE TABLE，不得改动块内 DDL 文本， +/// 并有 `tests::test_schema_sql_table_order_has_forward_reference_returns_none` 守门。 const SCHEMA_SQL: &str = r#" CREATE TABLE IF NOT EXISTS organizations ( id BIGINT AUTO_INCREMENT PRIMARY KEY, @@ -39,6 +47,26 @@ const SCHEMA_SQL: &str = r#" FOREIGN KEY (project_id) REFERENCES projects(id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + CREATE TABLE IF NOT EXISTS cabin_boards ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + cabinet_id BIGINT NOT NULL, + board_id TINYINT NOT NULL COMMENT '仓控板ID 1-6', + status TINYINT DEFAULT 0, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + + CREATE TABLE IF NOT EXISTS compartments ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + cabin_board_id BIGINT NOT NULL, + channel_id TINYINT NOT NULL COMMENT '通道ID 1-6', + status TINYINT DEFAULT 0 COMMENT '0空闲 1充电中 2故障 3禁用', + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + FOREIGN KEY (cabin_board_id) REFERENCES cabin_boards(id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + CREATE TABLE IF NOT EXISTS sim_change_logs ( id BIGINT AUTO_INCREMENT PRIMARY KEY, imei VARCHAR(15) NOT NULL COMMENT '设备IMEI', @@ -86,26 +114,6 @@ const SCHEMA_SQL: &str = r#" INDEX idx_compartment (compartment_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - CREATE TABLE IF NOT EXISTS cabin_boards ( - id BIGINT AUTO_INCREMENT PRIMARY KEY, - cabinet_id BIGINT NOT NULL, - board_id TINYINT NOT NULL COMMENT '仓控板ID 1-6', - status TINYINT DEFAULT 0, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - FOREIGN KEY (cabinet_id) REFERENCES cabinets(id) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - - CREATE TABLE IF NOT EXISTS compartments ( - id BIGINT AUTO_INCREMENT PRIMARY KEY, - cabin_board_id BIGINT NOT NULL, - channel_id TINYINT NOT NULL COMMENT '通道ID 1-6', - status TINYINT DEFAULT 0 COMMENT '0空闲 1充电中 2故障 3禁用', - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - FOREIGN KEY (cabin_board_id) REFERENCES cabin_boards(id) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - CREATE TABLE IF NOT EXISTS users ( id BIGINT AUTO_INCREMENT PRIMARY KEY, phone VARCHAR(11) NOT NULL UNIQUE COMMENT '手机号', @@ -246,8 +254,21 @@ pub async fn run(pool: &MySqlPool) -> Result<(), sqlx::Error> { } // 兼容已有数据库：将 abstract_id 改为可空 - let _ = sqlx::query(r#"ALTER TABLE cabinets MODIFY abstract_id VARCHAR(20) NULL"#) - .execute(pool).await; + // 只在列当前确实为 NOT NULL（老库）时才 ALTER：避免每次重启都对 cabinets 做一次 DDL + // 容错口径与修改前一致（原先为 `let _ =`）：失败只记 ERROR，绝不中断启动 + if let Err(e) = ensure_column_nullable( + pool, + "cabinets", + "abstract_id", + r#"ALTER TABLE cabinets MODIFY abstract_id VARCHAR(20) NULL"#, + ) + .await + { + tracing::error!( + "[migrate] cabinets.abstract_id 可空化迁移失败（服务继续启动）err={}", + e + ); + } // 更新 charge_records.max_temperature 注释（0.1°C → K×100） // 只在注释确实过时（老库）时才 ALTER：避免每次启动都对整表做一次 MODIFY @@ -468,6 +489,42 @@ async fn ensure_column( Ok(()) } +/// 列可空性幂等迁移：仅当该列**当前为 NOT NULL**（老库）时才 ALTER 为可空 +/// +/// 为什么：`cabinets.abstract_id` 在老库上是 `NOT NULL`，需要改成可空； +/// 但无条件执行 MODIFY 会让**每次服务启动**都对 `cabinets` 做一次 DDL +/// （MODIFY 会走「重建整表」路径，大表上会长时间持有元数据锁、阻塞并发读写）。 +/// 而该 DDL 只在老库上真正必要，新库/已迁移库应当完全跳过。 +/// +/// 安全性：表名/列名/DDL 全部来自本文件内的白名单常量，不含任何外部输入拼接。 +async fn ensure_column_nullable( + pool: &MySqlPool, + table: &str, + column: &str, + ddl: &str, +) -> Result<(), sqlx::Error> { + // ⚠️ 与 `ensure_column_comment` 同样的坑：MySQL 8 上 `information_schema` 的字符列 + // 会被 sqlx 报告为 **VARBINARY**，直接按 `Option` 解码会 panic， + // 必须显式 `CAST(... AS CHAR)`。 + let is_nullable: Option = sqlx::query_scalar( + "SELECT CAST(IS_NULLABLE AS CHAR) FROM information_schema.COLUMNS \ + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?", + ) + .bind(table) + .bind(column) + .fetch_optional(pool) + .await?; + + // 列不存在（None）时不动：建列由 ensure_column 负责 + if let Some(flag) = is_nullable { + if flag != "YES" { + sqlx::query(ddl).execute(pool).await?; + tracing::info!("[migrate] 已将 {}.{} 改为可空", table, column); + } + } + Ok(()) +} + /// 列注释幂等更新：仅当现有注释不包含 `marker` 时才 ALTER /// /// 为什么：老库的列注释不会随 CREATE TABLE IF NOT EXISTS 更新； @@ -765,3 +822,85 @@ async fn grant_permissions_by_codes( Ok(()) } + +#[cfg(test)] +mod tests { + use super::SCHEMA_SQL; + + /// `SCHEMA_SQL` 中每条建表语句的固定前缀 + const CREATE_PREFIX: &str = "CREATE TABLE IF NOT EXISTS "; + /// 外键引用的关键字（用于从建表语句中抽取被引用表名） + const REFERENCES_MARKER: &str = "REFERENCES "; + /// 期望的建表数量（表数量变化时该断言会提醒同步更新） + const EXPECTED_TABLE_COUNT: usize = 21; + + /// 抽取建表语句中所有 `REFERENCES <表名>` 的被引用表名 + fn referenced_tables(statement: &str) -> Vec<&str> { + let mut refs = Vec::new(); + let mut rest = statement; + while let Some(pos) = rest.find(REFERENCES_MARKER) { + let after = &rest[pos + REFERENCES_MARKER.len()..]; + let name = after + .split(|c: char| !(c.is_alphanumeric() || c == '_')) + .next() + .unwrap_or_default(); + if !name.is_empty() { + refs.push(name); + } + rest = after; + } + refs + } + + /// 按出现顺序解析 `SCHEMA_SQL`，返回 (表名, 建表语句全文) + fn parse_table_definitions() -> Vec<(String, &'static str)> { + let mut defs = Vec::new(); + for statement in SCHEMA_SQL.split(';') { + let statement = statement.trim(); + if let Some(rest) = statement.strip_prefix(CREATE_PREFIX) { + let name = rest + .split(|c: char| c.is_whitespace() || c == '(') + .next() + .unwrap_or_default() + .to_string(); + defs.push((name, statement)); + } + } + defs + } + + /// `SCHEMA_SQL` 必须按「被引用表在前」排列 —— + /// 脚本按 `;` 逐条执行，前向引用会让全新空库直接报 1824 并 panic 退出。 + #[test] + fn test_schema_sql_table_order_has_forward_reference_returns_none() { + let defs = parse_table_definitions(); + assert_eq!( + defs.len(), + EXPECTED_TABLE_COUNT, + "SCHEMA_SQL 建表数量与预期不符（新增/删除表时请同步更新本断言）" + ); + + let mut defined: Vec<&str> = Vec::new(); + for (table, statement) in &defs { + for referenced in referenced_tables(statement) { + assert!( + defined.contains(&referenced), + "前向引用：表 `{}` 引用了尚未创建的 `{}`（全新库执行会报 1824 \ + Failed to open the referenced table）", + table, + referenced + ); + } + defined.push(table); + } + } + + /// 解析器自检：`REFERENCES x(id)` 应被识别为表 `x` + #[test] + fn test_referenced_tables_single_foreign_key_returns_table_name() { + let refs = referenced_tables( + "CREATE TABLE IF NOT EXISTS t (a BIGINT, FOREIGN KEY (a) REFERENCES other(id))", + ); + assert_eq!(refs, vec!["other"]); + } +} diff --git a/software/api-server/src/routes/h5.rs b/software/api-server/src/routes/h5.rs index 48b8c00..d90bed4 100644 --- a/software/api-server/src/routes/h5.rs +++ b/software/api-server/src/routes/h5.rs @@ -1190,6 +1190,11 @@ pub async fn scan_cabinet( /// 协议 §5.7：**`on` 是建立订单的唯一入口**。平台指令模式（`work_mode="0"`）下由**平台**下发 `on`， /// 因此平台必须在**下发前**建单 —— 设备随后回的 `on` 只是回执（带 `suc`），回执不建单。 /// 建单失败只记日志、**绝不影响指令下发**（充电优先于订单统计）。 +/// +/// ⚠️ **校验顺序（本 handler 的契约）**： +/// **权限 → 归属 → 设备启用 → 设备在线 → 建单 → 下发**。 +/// 建单必须排在全部校验之后：否则设备离线/无权访问时接口虽然报错，订单却已经插进 +/// `charge_orders`（幽灵单，只能等 10 小时超时结单，UI 上还会多出一次"充电中"）。 pub async fn start_charge( user: CurrentUser, State(state): State, @@ -1197,9 +1202,9 @@ pub async fn start_charge( ) -> Result, AppError> { auth::check_permission(&user, "charge:start")?; - // 下发前建单：一次查出柜子/仓控板/通道 + 充电工作模式 + // 一次查出柜子/仓控板/通道 + 充电工作模式 // （`cabinets.work_mode` 是 TINYINT：0=平台指令模式 / 1=自动充电模式 / NULL 按 0 处理） - let order_row: Result)>, sqlx::Error> = sqlx::query_as( + let target: Option<(i64, String, i8, i8, Option)> = sqlx::query_as( r#"SELECT cab.id, cab.imei, cb.board_id, comp.channel_id, cab.work_mode FROM cabinets cab JOIN cabin_boards cb ON cb.cabinet_id = cab.id @@ -1208,62 +1213,88 @@ pub async fn start_charge( ) .bind(body.compartment_id) .fetch_optional(&state.mysql) - .await; + .await?; - match order_row { - // 只有**平台指令模式**（`work_mode="0"`/NULL）才由平台下发 `on` 并在此建单； - // **自动充电模式**（`"1"`）下订单由设备上报的 `on`（不带 `suc`）建立，平台不重复建单 - Ok(Some((cabinet_id, imei, board_id, channel_id, mode))) if mode.unwrap_or(0) == 0 => { - // 起始 SOC 与电池 id 取该通道最近一次 `bat_in` 的登记（`on` 报文本身不带 `info`） - match crate::workers::report_worker::load_channel_battery( - &state.redis, &imei, board_id, channel_id, - ) - .await - { - Some(battery) => { - if let Err(e) = crate::workers::report_worker::ensure_charge_order( - &state.mysql, - &state.redis, - &imei, - cabinet_id, - board_id, - channel_id, - &battery.battery_id, - Some(battery.soc), - ) - .await - { - tracing::error!( - "[h5] 下发 on 前建单失败 compartment_id={} err={}", - body.compartment_id, - e - ); - } - } - None => tracing::warn!( - "[h5] 该通道无 bat_in 登记，跳过建单 compartment_id={}", - body.compartment_id - ), - } - } - Ok(Some(_)) => tracing::info!( + // 仓体不存在：与 [`send_compartment_command`] 的报错语义一致（同一 JOIN 条件） + let (cabinet_id, imei, board_id, channel_id, mode) = + target.ok_or_else(|| AppError::NotFound("仓体对应的设备不存在".into()))?; + + // 归属 → 设备启用 → 设备在线：全部**提前到建单之前**（顺序与 [`send_compartment_command`] 一致， + // 校验函数语义未改；下发路径仍会再校验一次，幂等无副作用） + verify_cabinet_access(&user, cabinet_id, &state.mysql).await?; + ensure_cabinet_enabled(&state.mysql, cabinet_id).await?; + ensure_device_online(&state, &imei).await?; + + // 只有**平台指令模式**（`work_mode="0"`/NULL）才由平台下发 `on` 并在此建单； + // **自动充电模式**（`"1"`）下订单由设备上报的 `on`（不带 `suc`）建立，平台不重复建单 + if mode.unwrap_or(0) == 0 { + create_order_before_on(&state, &body, cabinet_id, &imei, board_id, channel_id).await; + } else { + tracing::info!( "[h5] 自动充电模式（work_mode=1），建单由设备上报的 on 负责 compartment_id={}", body.compartment_id - ), - Ok(None) => tracing::warn!( - "[h5] 下发 on 前跳过建单：仓体不存在 compartment_id={}", - body.compartment_id - ), - Err(e) => tracing::error!( - "[h5] 下发 on 前跳过建单：查询仓体出错 compartment_id={} err={}", - body.compartment_id, - e - ), + ); } send_compartment_command(&state, &user, body.compartment_id, "on").await } +/// 平台下发 `on` **之前**建单（失败只记日志，绝不影响指令下发）。 +/// +/// 电池 id 与起始 SOC 取该通道最近一次 `bat_in` 的登记（`on` 报文本身不带 `info`）。 +/// ⚠️ 登记键 `device:{imei}:board:{b}:ch:{c}:bat` 的 TTL 只有 48 小时 +/// （[`crate::workers::report_worker`] 的 `CHANNEL_BAT_TTL_SECS`）—— 集中充电柜里电池常在仓内 +/// 放置超过 48 小时，登记过期后**必须仍按"未知电池"建单**（与设备侧 `handle_device_on` 的兜底 +/// 口径一致）：否则整段会话没有 `charge_orders` / `charge_records` / 电量， +/// 用户在"充电记录"里看不到这次充电，属于**平台建单漏单**。 +async fn create_order_before_on( + state: &AppState, + body: &CompartmentActionInput, + cabinet_id: i64, + imei: &str, + board_id: i8, + channel_id: i8, +) { + let (battery_id, start_soc) = match crate::workers::report_worker::load_channel_battery( + &state.redis, + imei, + board_id, + channel_id, + ) + .await + { + Some(battery) => (battery.battery_id, Some(battery.soc)), + None => { + tracing::warn!( + "[h5] 该通道无 bat_in 登记，按未知电池建单 compartment_id={}", + body.compartment_id + ); + (String::new(), None) + } + }; + + // `ensure_charge_order` 内部自带「按仓体进程内锁 + 条件插入 + 唯一索引」， + // 重复下发 / 双击不会多建单 + if let Err(e) = crate::workers::report_worker::ensure_charge_order( + &state.mysql, + &state.redis, + imei, + cabinet_id, + board_id, + channel_id, + &battery_id, + start_soc, + ) + .await + { + tracing::error!( + "[h5] 下发 on 前建单失败 compartment_id={} err={}", + body.compartment_id, + e + ); + } +} + /// POST /api/h5/charge/stop — 停止充电 pub async fn stop_charge( user: CurrentUser, diff --git a/software/api-server/src/routes/users.rs b/software/api-server/src/routes/users.rs index e15f5d3..a604626 100644 --- a/software/api-server/src/routes/users.rs +++ b/software/api-server/src/routes/users.rs @@ -589,15 +589,7 @@ pub async fn update_user( let pool = &state.mysql; - // 检查用户存在 - let old: Option<(String, Option, i32, Option)> = sqlx::query_as( - "SELECT phone, name, role, organization_id FROM users WHERE id = ?" - ) - .bind(user_id) - .fetch_optional(pool) - .await?; - let (phone, old_name, old_role, old_org_id) = old - .ok_or_else(|| AppError::NotFound("用户不存在".into()))?; + let (phone, old_name, old_role, old_org_id) = load_user_for_update(pool, user_id).await?; let new_name = input.name.filter(|s| !s.trim().is_empty()).map(|s| s.trim().to_string()); let new_role = input.role.unwrap_or(old_role); @@ -616,10 +608,57 @@ pub async fn update_user( ) .await?; + persist_user_update(pool, user_id, &new_name, new_role, new_org_id, old_role).await?; + + log_user_update( + pool, + current_user.user_id, + user_id, + &phone, + old_name.as_deref(), + new_name.as_deref(), + old_role, + new_role, + ) + .await; + + Ok(Json(json!({ "code": 0, "data": null, "message": "更新成功" }))) +} + +/// 读取待编辑用户的原始信息（`phone` / `name` / `role` / `organization_id`） +/// +/// 不存在时直接返回 `NotFound`。原属 [`update_user`] 的"检查用户存在"段。 +async fn load_user_for_update( + pool: &sqlx::MySqlPool, + user_id: i64, +) -> Result<(String, Option, i32, Option), AppError> { + // 检查用户存在 + let old: Option<(String, Option, i32, Option)> = sqlx::query_as( + "SELECT phone, name, role, organization_id FROM users WHERE id = ?" + ) + .bind(user_id) + .fetch_optional(pool) + .await?; + let (phone, old_name, old_role, old_org_id) = old + .ok_or_else(|| AppError::NotFound("用户不存在".into()))?; + Ok((phone, old_name, old_role, old_org_id)) +} + +/// 落库用户改名/改角色/改组织，并在**角色真的变了**时同步 `user_roles` 映射 +/// +/// 原属 [`update_user`] 的 `UPDATE` 段与角色同步段（注释随之原样搬移）。 +async fn persist_user_update( + pool: &sqlx::MySqlPool, + user_id: i64, + new_name: &Option, + new_role: i32, + new_org_id: Option, + old_role: i32, +) -> Result<(), AppError> { let rows = sqlx::query( "UPDATE users SET name = ?, role = ?, organization_id = ? WHERE id = ?" ) - .bind(&new_name) + .bind(new_name) .bind(new_role) .bind(new_org_id) .bind(user_id) @@ -646,19 +685,33 @@ pub async fn update_user( sync_user_role_mapping(pool, user_id, new_role).await?; } + Ok(()) +} + +/// 记录"编辑用户"操作日志（含变更前后对比）并打业务日志 +/// +/// 原属 [`update_user`] 的"记录操作日志"段。 +async fn log_user_update( + pool: &sqlx::MySqlPool, + actor_id: i64, + user_id: i64, + phone: &str, + old_name: Option<&str>, + new_name: Option<&str>, + old_role: i32, + new_role: i32, +) { // 记录操作日志 let detail = format!( "编辑用户: 手机号={}, 姓名: {}→{}, 角色: {}→{}", phone, - old_name.as_deref().unwrap_or("-"), - new_name.as_deref().unwrap_or("-"), + old_name.unwrap_or("-"), + new_name.unwrap_or("-"), old_role, new_role ); - log_operation(pool, current_user.user_id, "edit", "user", Some(user_id), &detail).await; + log_operation(pool, actor_id, "edit", "user", Some(user_id), &detail).await; tracing::info!(user_id, "编辑用户"); - - Ok(Json(json!({ "code": 0, "data": null, "message": "更新成功" }))) } // --------------------------------------------------------------------------- diff --git a/software/api-server/src/workers/report_worker.rs b/software/api-server/src/workers/report_worker.rs index b9b7024..eb98823 100644 --- a/software/api-server/src/workers/report_worker.rs +++ b/software/api-server/src/workers/report_worker.rs @@ -53,6 +53,12 @@ const REDIS_POLL_INTERVAL_MS: u64 = 200; /// Redis key 前缀 const REDIS_KEY_AUTH_STR: &str = "auth_str:"; + +/// Redis 队列深度 INFO 日志的间隔（秒）。 +/// +/// 取 10 分钟：只为在日志里留下**积压趋势**（`device:reports` / `device:commands` / +/// `device:commands:dead`），频率低到对 Redis 无感（3 次 `LLEN` / 10 分钟）。 +const QUEUE_DEPTH_LOG_INTERVAL_SECS: u64 = 600; /// 签名截取范围（8 hex = 4 bytes，受 MCU 硬件协议约束不可改） /// 审查编号：I5（已确认不修，硬件约束。设备一次只能获取一次 auth_str，除非显式标记重新生成） const SIGN_START: usize = 56; @@ -96,6 +102,17 @@ const MAX_CHANNEL_ID: i64 = 8; /// 占用状态对账周期：与 `order_timeout` 的订单超时扫描同频（300 秒），避免额外数据库压力 const RECONCILE_INTERVAL_SECS: u64 = 300; +/// 「无订单却标占用」兜底复位的**陈旧阈值**（小时）。 +/// +/// 依据：`order_timeout` 的订单超时是 **10 小时**（见 [`ORDER_TIMEOUT_HOURS`]）， +/// 任何**真实存在过的**充电会话最迟 10 小时就会被结单、从而落入"有已结束订单"的正常复位分支。 +/// 因此一条 `status=1` **且一笔订单都没有**、又**超过该阈值没有更新**的仓体， +/// 不可能是正在充电的会话，只能是历史遗留的错误状态（例如平台建单漏单时代留下的脏标记）。 +/// +/// 取 **12 小时 = 10 小时订单超时 + 2 小时余量**：宁可多留 2 小时也不早复位。 +/// ⚠️ 该阈值只是**拿不到 Redis 通道状态时**的兜底；优先判据是通道 `batt_status` 的 bit7（有无电池）。 +const RECONCILE_ORDERLESS_STALE_HOURS: i64 = 12; + // --------------------------------------------------------------------------- // 频率限制器 // --------------------------------------------------------------------------- @@ -139,6 +156,9 @@ pub async fn start(redis: ConnectionManager, mysql: MySqlPool) { // 靠"通道连续报无电池"提前结单（待确认第 18 项的兜底）—— 按分工这都属于本 worker 的职责。 tokio::spawn(reconcile_usage_loop(mysql.clone(), redis.clone())); + // 低频队列深度日志（见 [`QUEUE_DEPTH_LOG_INTERVAL_SECS`]）：让"积压"在日志里可见 + let mut last_depth_log = Instant::now(); + loop { let mut conn = redis.clone(); // 非阻塞轮询取一条上报（原因见 [`REDIS_POLL_INTERVAL_MS`]）： @@ -165,6 +185,30 @@ pub async fn start(redis: ConnectionManager, mysql: MySqlPool) { tokio::time::sleep(Duration::from_secs(1)).await; } } + + // 队列深度周期性 INFO 日志（节流，仅观测） + if last_depth_log.elapsed() >= Duration::from_secs(QUEUE_DEPTH_LOG_INTERVAL_SECS) { + last_depth_log = Instant::now(); + log_queue_depths(&redis).await; + } + } +} + +/// 周期性记录关键 Redis 队列深度（INFO；读取失败只记 WARN，绝不影响消费）。 +/// +/// 为什么需要：Redis 实例 `maxmemory=1GB` + `maxmemory-policy=volatile-lru`，而实例上绝大多数 +/// key 没有 TTL（可淘汰集合极小）—— 一旦 `device:reports` 无界积压，写命令 OOM 会**连累同实例上 +/// 既有的充电站生产平台**，而积压本身只能从队列深度看出来（消费停滞时没有错误日志）。 +/// 生产端已加 `LTRIM` 上限（见 `device-server/src/tcp/handler.rs`），这里补上"看得见"的一环。 +async fn log_queue_depths(redis: &ConnectionManager) { + for key in ["device:reports", "device:commands", "device:commands:dead"] { + let mut conn = redis.clone(); + let depth: Result = + redis::cmd("LLEN").arg(key).query_async(&mut conn).await; + match depth { + Ok(d) => tracing::info!("[queue] {} depth={}", key, d), + Err(e) => tracing::warn!("[queue] 读取队列深度失败 key={} err={}", key, e), + } } } @@ -1534,12 +1578,10 @@ pub async fn ensure_charge_order( battery_id: &str, start_soc: Option, ) -> Result<(), Box> { - let compartment_id = match find_compartment_id(pool, cabinet_id, board_id, channel_id).await? { + let compartment_id = match resolve_order_compartment(pool, cabinet_id, board_id, channel_id).await? + { Some(id) => id, - None => { - tracing::warn!("[order] 仓体不存在，跳过建单 cabinet_id={} board={} ch={}", cabinet_id, board_id, channel_id); - return Ok(()); - } + None => return Ok(()), }; // 「查活跃单 → 判定 → 结束老单 → 插入新单」**整段按仓体串行化**（进程内锁，见 @@ -1553,36 +1595,18 @@ pub async fn ensure_charge_order( // 锁只护住本进程、条件插入在"读快照"实现里仍有窗口、唯一键只能兜底报错。 let _guard = acquire_compartment_lock(compartment_id).await; - let active = find_active_order(pool, compartment_id).await?; - - // 是否需要"换电池"处理：两边电池 id 都已知且不同（任一为空视为未变） - let decision = decide_order_action(active.as_ref().map(|(_, b)| b.as_str()), battery_id); - - match &active { - None => {} - Some(_) if decision == OrderDecision::Skip => { - tracing::debug!( - "[order] 通道已有活跃订单且电池未变，不新建 compartment_id={}", - compartment_id - ); - return Ok(()); - } - Some((order_id, old_battery)) => { - // 一个通道只可能有一块电池：id 变了必然是换了新电池，老订单必须先结束。 - // 走统一结算：写回本次充电电量（平台侧积分）并生成充电记录。 - end_replaced_order( - pool, - redis, - imei, - board_id, - channel_id, - compartment_id, - battery_id, - *order_id, - old_battery, - ) - .await?; - } + if !reconcile_active_order( + pool, + redis, + imei, + board_id, + channel_id, + compartment_id, + battery_id, + ) + .await? + { + return Ok(()); } let start_soc = match start_soc { @@ -1613,6 +1637,73 @@ pub async fn ensure_charge_order( Ok(()) } +/// 解析建单目标仓体；仓体不存在时告警并返回 `None`（调用方据此跳过建单） +/// +/// 原属 [`ensure_charge_order`] 的"定位仓体"段。 +async fn resolve_order_compartment( + pool: &MySqlPool, + cabinet_id: i64, + board_id: i8, + channel_id: i8, +) -> Result, Box> { + match find_compartment_id(pool, cabinet_id, board_id, channel_id).await? { + Some(id) => Ok(Some(id)), + None => { + tracing::warn!("[order] 仓体不存在，跳过建单 cabinet_id={} board={} ch={}", cabinet_id, board_id, channel_id); + Ok(None) + } + } +} + +/// 建单前的活跃单对账：已存在且电池未变 ⇒ 返回 `false`（跳过建单）； +/// 电池变了 ⇒ 先结束老订单再返回 `true` +/// +/// 原属 [`ensure_charge_order`] 的"查活跃单 → 判定 → 结束老单"段， +/// 调用方必须已持有该仓体的进程内锁（见 [`acquire_compartment_lock`]）。 +async fn reconcile_active_order( + pool: &MySqlPool, + redis: &ConnectionManager, + imei: &str, + board_id: i8, + channel_id: i8, + compartment_id: i64, + battery_id: &str, +) -> Result> { + let active = find_active_order(pool, compartment_id).await?; + + // 是否需要"换电池"处理：两边电池 id 都已知且不同（任一为空视为未变） + let decision = decide_order_action(active.as_ref().map(|(_, b)| b.as_str()), battery_id); + + match &active { + None => {} + Some(_) if decision == OrderDecision::Skip => { + tracing::debug!( + "[order] 通道已有活跃订单且电池未变，不新建 compartment_id={}", + compartment_id + ); + return Ok(false); + } + Some((order_id, old_battery)) => { + // 一个通道只可能有一块电池：id 变了必然是换了新电池，老订单必须先结束。 + // 走统一结算：写回本次充电电量（平台侧积分）并生成充电记录。 + end_replaced_order( + pool, + redis, + imei, + board_id, + channel_id, + compartment_id, + battery_id, + *order_id, + old_battery, + ) + .await?; + } + } + + Ok(true) +} + /// 查该仓体当前活跃订单（`status = 0`）→ `(订单 id, 电池 id)`；无活跃订单返回 `None`。 async fn find_active_order( pool: &MySqlPool, @@ -2043,6 +2134,8 @@ async fn mark_battery_status(pool: &MySqlPool, battery_id: &str, status: i8) { /// 走到 `bat_out` / `off` 的复位分支，若不兜底，仓体会永远显示"充电中"。 /// 为什么加 `EXISTS(已结束订单)` 条件：**刚 `bat_in` 还未开始充电**的仓体/电池 /// （`status=1` 但一笔订单都没有）不应被误复位，那属于正常的"已插入、待充电"状态。 +/// ⚠️ 该条件也意味着"零订单的历史脏行"永不自愈 —— 这类行由 +/// [`reconcile_orderless_compartments`] 用**通道状态 + 陈旧阈值**单独兜底（见该函数说明）。 async fn reconcile_usage_loop(mysql: MySqlPool, redis: ConnectionManager) { let mut ticker = tokio::time::interval(Duration::from_secs(RECONCILE_INTERVAL_SECS)); loop { @@ -2055,9 +2148,11 @@ async fn reconcile_usage_loop(mysql: MySqlPool, redis: ConnectionManager) { /// 单轮对账（见 [`reconcile_usage_loop`] 的说明） /// -/// 两件事： +/// 三件事： /// 1. **占用状态复位**：订单已全部结束但 `compartments`/`batteries` 还标着"占用"的，置回空闲 -/// 2. **"无电池"状态兜底**（**待确认第 18 项**）：该通道 `batt_status` 的 **bit7 连续 +/// 2. **"无订单却标占用"兜底复位**（[`reconcile_orderless_compartments`]）：`status=1` 但 +/// **一笔订单都没有**的行走不到第 1 步（它要求 `EXISTS(已结束订单)`），必须单独兜底 +/// 3. **"无电池"状态兜底**（**待确认第 18 项**）：该通道 `batt_status` 的 **bit7 连续 /// [`energy::NO_BATTERY_STRIKES`] 次为 0**（协议 §5.8 位定义：0=无电池）却还挂着活跃订单 /// ⇒ 判定"人把电池拔走了但设备没机会上报 `bat_out`"，**提前结单**，避免订单一直挂到 10 小时超时。 /// @@ -2098,9 +2193,141 @@ async fn reconcile_usage_once( batteries.rows_affected() ); } + + reconcile_orderless_compartments(mysql, redis).await?; Ok(()) } +/// 兜底复位「`compartments.status=1` 但**一笔订单都没有**」的行。 +/// +/// **为什么必须有这一步（生产实测事故）**：`reconcile_usage_once` 的主复位要求 +/// `EXISTS(已结束订单)`，而"平台下发 `on` 前建单失败/漏单"时代留下的行**一笔订单都没有**， +/// 于是永不自愈 —— 生产库 936 个仓体里曾长期有 **214 个 `status=1`（充电中）却零订单**， +/// UI 与统计口径都跟着错（最后更新停在很久以前）。新增条件/新增语句，**不改既有过滤语义**。 +/// +/// 判据（先严后宽，只清"确实空闲"的行）： +/// 1. 无任何订单（`NOT EXISTS`）—— 有活跃订单必然保留；有已结束订单的交给主复位分支； +/// 2. **优先**用 Redis 通道状态 `batt_status` 的 **bit7**（协议 §5.8：bit7=1 有电池 / 0 无电池）： +/// - bit7=0 ⇒ 通道里没有电池 ⇒ 确实空闲 ⇒ 复位； +/// - bit7=1 ⇒ 电池还在仓里（可能是"已插入待充电"的正常状态）⇒ **保留**； +/// 3. 拿不到通道状态（键不存在 / Redis 故障 / Redis 重启后状态丢失）⇒ 退回 +/// "最后更新时间超过 [`RECONCILE_ORDERLESS_STALE_HOURS`] 小时"作为兜底（依据见该常量）。 +/// +/// 复位按**单行 + 同样的守卫条件**执行（`UPDATE ... WHERE id=? AND status=1 AND NOT EXISTS(...)`）， +/// 避免"扫描与复位之间刚 `bat_in` 建了单"被误复位。 +async fn reconcile_orderless_compartments( + mysql: &MySqlPool, + redis: &ConnectionManager, +) -> Result<(), sqlx::Error> { + // 候选行：`status=1` 且该仓体**没有任何订单**；顺带查出通道定位信息与"是否已陈旧" + let candidates: Vec<(i64, String, i8, i8, bool)> = sqlx::query_as( + "SELECT comp.id, cab.imei, cb.board_id, comp.channel_id, + (comp.updated_at < (NOW() - INTERVAL ? HOUR)) AS stale + FROM compartments comp + JOIN cabin_boards cb ON cb.id = comp.cabin_board_id + JOIN cabinets cab ON cab.id = cb.cabinet_id + WHERE comp.status = 1 + AND NOT EXISTS (SELECT 1 FROM charge_orders co WHERE co.compartment_id = comp.id) + LIMIT 500", + ) + .bind(RECONCILE_ORDERLESS_STALE_HOURS) + .fetch_all(mysql) + .await?; + + let scanned = candidates.len(); + // 复位计数按判据分开统计，便于运维从日志区分"通道确认空闲"与"陈旧兜底" + let mut reset_idle = 0u64; + let mut reset_stale = 0u64; + let mut kept_battery_present = 0u64; + + for (compartment_id, imei, board_id, channel_id, stale) in candidates { + match channel_battery_present(redis, &imei, board_id.into(), channel_id.into()).await { + // 通道状态可用：bit7=0（无电池）⇒ 确实空闲 + Some(false) => { + reset_idle += reset_orderless_compartment(mysql, compartment_id).await?; + } + // bit7=1 ⇒ 电池还在仓里，可能是"已插入待充电"，**不动** + Some(true) => kept_battery_present += 1, + // 拿不到状态 ⇒ 只能按"最后更新时间超过 N 小时"兜底 + None if stale => { + reset_stale += reset_orderless_compartment(mysql, compartment_id).await?; + } + None => {} + } + } + + if reset_idle > 0 || reset_stale > 0 { + tracing::warn!( + "[reconcile] 无订单却标占用（status=1）的仓体已兜底复位：\ + 按通道状态 batt_status.bit7=0（确实无电池）复位 {} 行，\ + 按最后更新时间超过 {} 小时（拿不到通道状态）复位 {} 行；\ + 本轮候选 {} 行，因通道仍报有电池而保留 {} 行", + reset_idle, + RECONCILE_ORDERLESS_STALE_HOURS, + reset_stale, + scanned, + kept_battery_present + ); + } else if scanned > 0 && kept_battery_present > 0 { + tracing::info!( + "[reconcile] 无订单却标占用（status=1）的仓体 {} 行均报有电池，按「已插入待充电」保留", + kept_battery_present + ); + } + + Ok(()) +} + +/// 读取通道 `batt_status` 的 **bit7**（有无电池）：`Some(true)`=有电池 / `Some(false)`=无电池 / +/// `None`=拿不到状态（键不存在或 Redis 读取失败）。 +/// +/// 位定义见协议 §5.8 与 [`energy::accumulate`] 的同一判据（`batt_status & 0x80`）。 +/// 键为无 TTL 的通道实时状态 Hash（`report_worker` 每次 `status_post` 覆盖）， +/// 因此"键不存在"只可能是设备长期离线或 Redis 重启 —— 调用方需自行兜底。 +async fn channel_battery_present( + redis: &ConnectionManager, + imei: &str, + board_id: i64, + channel_id: i64, +) -> Option { + let key = channel_status_key(imei, board_id, channel_id); + let mut conn = redis.clone(); + let value: Result, redis::RedisError> = redis::cmd("HGET") + .arg(&key) + .arg("batt_status") + .query_async(&mut conn) + .await; + + match value { + Ok(Some(hex)) => u8::from_str_radix(hex.trim(), 16).ok().map(|b| b & 0x80 != 0), + Ok(None) => None, + Err(e) => { + tracing::warn!("[reconcile] 读取通道状态失败 key={} err={}", key, e); + None + } + } +} + +/// 复位**单行**"无订单却标占用"的仓体，返回实际影响行数。 +/// +/// 守卫条件与候选查询完全一致（`status=1` + 无任何订单）， +/// 因此即便扫描与复位之间该仓体刚建了单，也**不会**被误复位。 +async fn reset_orderless_compartment( + mysql: &MySqlPool, + compartment_id: i64, +) -> Result { + let result = sqlx::query( + "UPDATE compartments SET status = 0 + WHERE id = ? AND status = 1 + AND NOT EXISTS (SELECT 1 FROM charge_orders co WHERE co.compartment_id = compartments.id)", + ) + .bind(compartment_id) + .execute(mysql) + .await?; + + Ok(result.rows_affected()) +} + /// **待确认第 18 项的兜底**：对"通道连续多次报无电池但订单还挂着"的订单提前结单。 /// /// 场景：停电/离线期间人把电池取走 —— 设备**根本没机会上报 `bat_out`**， @@ -2552,11 +2779,63 @@ async fn handle_bat_out( .map(|c| c.soc); // 结束该通道活跃订单（统一结算：写回平台侧积分电量 + 生成充电记录）。 + let updated = finalize_active_order_for_bat_out( + mysql, + redis, + dev_id, + board_id, + channel_id, + compartment_id, + battery_id, + end_soc, + ) + .await?; + + reset_usage_status_unless_finalized(mysql, dev_id, compartment_id, battery_id, updated).await; + + // 电池已离柜：清掉通道电池登记，避免下一次 `on` 用到已拔出的电池 id。 // - // 先判定"这条 bat_out 是不是当前这笔活跃订单的电池"（见 [`should_finalize_bat_out`]）： - // `bat_out` 可能是**上一块电池的迟到报文**（协议 §5.6 要求先发 off 再发 bat_out， - // 两条报文由 `tokio::spawn` 并发处理，中间人还可能已经换上新电池并建了新单）， - // 那种情况下按仓体收尾会把**正在充电的新单**结束掉（reason 记成 user_stop、电量错位）。 + // ⚠️ 这里**刻意不按 `updated > 0` 门控**（虽然占用复位是门控的）：`off` + `bat_out` + // 成对到达时，胜出的往往是 `off` ⇒ `bat_out` 拿到的 `updated == 0`， + // 而**清理通道登记只有 `bat_out` 这一条路径**，门控会让登记永远残留。 + // 真正的保护是"**登记的 battery_id 必须与本次 bat_out 的 battery_id 一致**"： + // 若该通道已经插入了新电池（新 `bat_in` 覆盖了登记），就不会误删新电池的登记。 + delete_channel_battery_if_matches(redis, dev_id, board_id, channel_id, battery_id).await; + + log_bat_out(mysql, dev_id, msg, compartment_id, battery_id, end_soc).await; + + tracing::info!( + "[bat_out] 电池拔出 dev_id={} compartment_id={} battery_id={} end_soc={:?}", + dev_id, + compartment_id, + battery_id, + end_soc + ); + + let resp = ServerResponse::success_for(msg); + send_response_to_device(redis, dev_id, &resp).await?; + + Ok(()) +} + +/// 结束 `bat_out` 对应的活跃订单，返回受影响行数（`0` = 没有可结束的活跃订单） +/// +/// 先判定"这条 bat_out 是不是当前这笔活跃订单的电池"（见 [`should_finalize_bat_out`]）： +/// `bat_out` 可能是**上一块电池的迟到报文**（协议 §5.6 要求先发 off 再发 bat_out， +/// 两条报文由 `tokio::spawn` 并发处理，中间人还可能已经换上新电池并建了新单）， +/// 那种情况下按仓体收尾会把**正在充电的新单**结束掉（reason 记成 user_stop、电量错位）。 +/// +/// 原属 [`handle_bat_out`] 的"结束活跃订单"段。 +async fn finalize_active_order_for_bat_out( + mysql: &MySqlPool, + redis: &ConnectionManager, + dev_id: &str, + board_id: i8, + channel_id: i8, + compartment_id: i64, + battery_id: &str, + end_soc: Option, +) -> Result> { let active = find_active_order(mysql, compartment_id).await?; let active_battery = active.as_ref().map(|(_, battery)| battery.as_str()); let updated = if should_finalize_bat_out(active_battery, battery_id) { @@ -2588,11 +2867,24 @@ async fn handle_bat_out( ); 0 }; + Ok(updated) +} +/// 订单确实结束（`updated > 0`）时才复位占用状态；否则告警且**不动**占用标记 +/// +/// 没有活跃订单可结束（重发 bat_out / 已被 off 或超时路径结单）⇒ +/// **不复位占用状态**（与 [`handle_device_off`] 一致）：此时仓体/电池可能已经属于 +/// **新的一单**（新电池插入 + `on` 建单），盲目复位会把新单的占用标记清掉。 +/// +/// 原属 [`handle_bat_out`] 的"占用状态复位"段。 +async fn reset_usage_status_unless_finalized( + mysql: &MySqlPool, + dev_id: &str, + compartment_id: i64, + battery_id: &str, + updated: u64, +) { if updated == 0 { - // 没有活跃订单可结束（重发 bat_out / 已被 off 或超时路径结单）⇒ - // **不复位占用状态**（与 [`handle_device_off`] 一致）：此时仓体/电池可能已经属于 - // **新的一单**（新电池插入 + `on` 建单），盲目复位会把新单的占用标记清掉。 tracing::warn!( "[bat_out] 无活跃订单可结束，不复位占用状态 dev_id={} compartment_id={}", dev_id, @@ -2602,30 +2894,6 @@ async fn handle_bat_out( // 订单结束 ⇒ 占用状态复位：仓体空闲（`batteries` 为空串时跳过台账） reset_usage_status(mysql, compartment_id, battery_id).await; } - - // 电池已离柜：清掉通道电池登记，避免下一次 `on` 用到已拔出的电池 id。 - // - // ⚠️ 这里**刻意不按 `updated > 0` 门控**（虽然占用复位是门控的）：`off` + `bat_out` - // 成对到达时，胜出的往往是 `off` ⇒ `bat_out` 拿到的 `updated == 0`， - // 而**清理通道登记只有 `bat_out` 这一条路径**，门控会让登记永远残留。 - // 真正的保护是"**登记的 battery_id 必须与本次 bat_out 的 battery_id 一致**"： - // 若该通道已经插入了新电池（新 `bat_in` 覆盖了登记），就不会误删新电池的登记。 - delete_channel_battery_if_matches(redis, dev_id, board_id, channel_id, battery_id).await; - - log_bat_out(mysql, dev_id, msg, compartment_id, battery_id, end_soc).await; - - tracing::info!( - "[bat_out] 电池拔出 dev_id={} compartment_id={} battery_id={} end_soc={:?}", - dev_id, - compartment_id, - battery_id, - end_soc - ); - - let resp = ServerResponse::success_for(msg); - send_response_to_device(redis, dev_id, &resp).await?; - - Ok(()) } /// 订单结束 ⇒ 占用状态复位：仓体置空闲 + 电池台账置空闲（`battery_id` 为空串时跳过台账）。 diff --git a/software/device-server/src/main.rs b/software/device-server/src/main.rs index 61f7429..eaffd5d 100644 --- a/software/device-server/src/main.rs +++ b/software/device-server/src/main.rs @@ -23,6 +23,30 @@ use config::Config; /// 设备侧不可见（与 `_close` 同类）。 const ROUTE_DEV_ID_FIELD: &str = "_dev_id"; +/// 死信队列 key（下发失败、不重试的指令）。 +/// +/// 历史上只写不读、**无 TTL、无上限、无告警**：设备长期离线时它会无限增长， +/// 在 `maxmemory-policy=volatile-lru` 且既有 key 几乎都无 TTL 的实例上就是一颗容量炸弹。 +const DEAD_LETTER_KEY: &str = "device:commands:dead"; + +/// 死信队列长度上限：保留**最近** 10000 条（更早的直接丢弃）。 +/// +/// 取 10000 的依据：死信是"无法下发"的指令，排查时只需要近期样本（单条指令 JSON 约 200~400B， +/// 10000 条 ≈ 2~4MB，对 1GB 实例可忽略）；数量级再大也无人逐条看，只会白占内存。 +const DEAD_LETTER_MAX: i64 = 10_000; + +/// 死信保留期：**7 天**（604800 秒）。 +/// +/// 依据：与 `log_retention` 的日志保留口径一致 —— 设备离线一周以上的历史指令已无排查价值， +/// 而 TTL 保证即使长时间无新死信写入，历史死信也不会永久占据内存。 +const DEAD_LETTER_TTL_SECS: i64 = 7 * 24 * 3600; + +/// Redis 队列深度 INFO 日志的间隔（秒）。 +/// +/// 取 10 分钟：只为在日志里留下**积压趋势**（`device:commands` / `device:reports` / +/// `device:commands:dead`），频率低到对 Redis 无感（每次 3~4 条 `LLEN`）。 +const QUEUE_DEPTH_LOG_INTERVAL_SECS: u64 = 600; + #[tokio::main] async fn main() { tracing_subscriber::fmt() @@ -144,6 +168,9 @@ async fn consume_commands( tracing::info!("[cmd_consumer] 启动 node_id={}", node_id); let my_queue = format!("device:commands:{}", node_id); + // 低频队列深度日志（见 [`QUEUE_DEPTH_LOG_INTERVAL_SECS`]）：让"积压"在日志里可见 + let mut last_depth_log = std::time::Instant::now(); + loop { let mut conn = redis.clone(); // **非阻塞轮询**取一条待下发指令，先取本节点专属队列、再取统一队列。 @@ -235,12 +262,8 @@ async fn consume_commands( Err(e) => { tracing::error!("[cmd] 发送失败 dev_id={} err={} 已入死信", dev_id, e); // ponytail: 丢入死信队列，不重试（重试语义和次数待业务明确） - let mut dl_conn = redis.clone(); - let _: Result<(), _> = redis::cmd("LPUSH") - .arg("device:commands:dead") - .arg(&cmd_json) - .query_async(&mut dl_conn) - .await; + // 入队同时**限长 + 设 TTL**，并回报累计死信数（可观测性） + push_dead_letter(&redis, &cmd_json, &dev_id, &e.to_string()).await; } } } @@ -256,5 +279,78 @@ async fn consume_commands( tokio::time::sleep(Duration::from_secs(1)).await; } } + + // 队列深度周期性 INFO 日志（节流，仅观测） + if last_depth_log.elapsed() >= Duration::from_secs(QUEUE_DEPTH_LOG_INTERVAL_SECS) { + last_depth_log = std::time::Instant::now(); + log_queue_depths(&redis, &my_queue).await; + } + } +} + +/// 写入死信队列（**限长 + TTL + WARN**），失败只记 ERROR、不影响消费循环。 +/// +/// - `LTRIM 0 DEAD_LETTER_MAX-1`：只保留最近 `DEAD_LETTER_MAX` 条； +/// - `EXPIRE DEAD_LETTER_TTL_SECS`：即使不再有死信写入，历史死信也会自然过期； +/// - `WARN` 里带上 `LLEN` 得到的**累计死信数**：让运维能从日志看出"死信在持续增长"， +/// 而不是只看到孤立的一条 ERROR（历史实现的问题正是"只写不读、无任何可观测性"）。 +async fn push_dead_letter( + redis: &redis::aio::ConnectionManager, + cmd_json: &str, + dev_id: &str, + err: &str, +) { + let mut conn = redis.clone(); + // 3 条命令 1 次往返：入队 → 限长 → 设 TTL（全部 `ignore()`，返回 `()`） + let mut pipe = redis::pipe(); + pipe.cmd("LPUSH").arg(DEAD_LETTER_KEY).arg(cmd_json).ignore(); + pipe.cmd("LTRIM") + .arg(DEAD_LETTER_KEY) + .arg(0) + .arg(DEAD_LETTER_MAX - 1) + .ignore(); + pipe.cmd("EXPIRE") + .arg(DEAD_LETTER_KEY) + .arg(DEAD_LETTER_TTL_SECS) + .ignore(); + + let pushed: Result<(), redis::RedisError> = pipe.query_async(&mut conn).await; + if let Err(e) = pushed { + tracing::error!("[cmd] 死信入队失败 dev_id={} err={}", dev_id, e); + return; + } + + // 再取一次长度作为"累计死信数"（读取失败不影响死信已入队的事实，按 -1 打日志） + let mut depth_conn = redis.clone(); + let depth: i64 = redis::cmd("LLEN") + .arg(DEAD_LETTER_KEY) + .query_async(&mut depth_conn) + .await + .unwrap_or(-1); + + tracing::warn!( + "[cmd] 指令已入死信队列 dev_id={} 累计死信数={} （上限 {} 条 / 保留 {} 秒）err={}", + dev_id, + depth, + DEAD_LETTER_MAX, + DEAD_LETTER_TTL_SECS, + err + ); +} + +/// 周期性记录关键 Redis 队列深度（INFO；读取失败只记 WARN，绝不影响消费）。 +/// +/// 为什么需要：Redis 实例 `maxmemory=1GB` + `maxmemory-policy=volatile-lru`，实例上绝大多数 key +/// 没有 TTL（可淘汰集合≈1）—— 队列无界积压会让写命令 OOM 并**连累同实例上既有充电站生产平台**； +/// 而消费端卡死时没有任何错误日志，深度日志是唯一的预警。 +async fn log_queue_depths(redis: &redis::aio::ConnectionManager, node_queue: &str) { + for key in ["device:commands", node_queue, "device:reports", DEAD_LETTER_KEY] { + let mut conn = redis.clone(); + let depth: Result = + redis::cmd("LLEN").arg(key).query_async(&mut conn).await; + match depth { + Ok(d) => tracing::info!("[queue] {} depth={}", key, d), + Err(e) => tracing::warn!("[queue] 读取队列深度失败 key={} err={}", key, e), + } } } diff --git a/software/device-server/src/tcp/handler.rs b/software/device-server/src/tcp/handler.rs index a232304..e87cce9 100644 --- a/software/device-server/src/tcp/handler.rs +++ b/software/device-server/src/tcp/handler.rs @@ -10,6 +10,24 @@ use super::connection::ConnectionPool; use super::protocol::DeviceMessage; +/// 上报队列 `device:reports` 的**长度上限**（超过就裁剪最旧的）。 +/// +/// **为什么必须有上限**：本服务只 `LPUSH`，api-server 是唯一消费者；一旦 api-server 长时间不消费 +/// （宕机/卡死/上线顺序颠倒），队列会无界增长，而 Redis 实例 `maxmemory=1GB`、 +/// `maxmemory-policy=volatile-lru` 且实例上绝大多数 key 没有 TTL（可淘汰集合≈1）⇒ +/// **写命令 OOM 会连累同实例上既有的充电站生产平台**。 +/// +/// **为什么取 200000**（估算依据）： +/// - 单条 `status_post` 序列化后约 **1.1~1.2KB**（含 36 通道 18 字符 hex + 柜级字段）； +/// - 68 柜规模：每柜每天 1440 条（协议 §5.3 固定 60 秒一次）⇒ 约 **9.8 万条/天 ≈ 115MB/天**， +/// 20 万条可容纳 **约 2 天**的完全积压（远超"重启窗口 + 正常积压"）； +/// - 1000 柜规模：约 **144 万条/天**，20 万条 ≈ **3.3 小时** 的缓冲； +/// - 内存占用上限 ≈ 20 万 × 1.15KB ≈ **230MB**，只占 1GB 实例的约 23%， +/// 给同实例其他业务留足余量（这正是设上限的目的）。 +/// +/// ⚠️ 不改队列名、不改报文格式（协议冻结）。 +const REPORTS_QUEUE_MAX: i64 = 200_000; + /// 分发消息：补写设备身份后转发到 Redis 上报队列。 /// /// `current_dev_id` = 本 TCP 连接绑定的设备身份（由 `auth_str`/`login` 建立）。 @@ -50,6 +68,11 @@ fn attach_dev_id(msg: &mut DeviceMessage, current_dev_id: Option<&str>) { } /// 将设备消息转发到 Redis 统一上报队列 `device:reports` +/// +/// `LPUSH` 后用同一条 pipeline 做 `LLEN → LTRIM → LLEN`： +/// - 保证队列**有界**（上限见 [`REPORTS_QUEUE_MAX`]）； +/// - 裁剪发生时记 **WARN** 并给出被丢弃条数 —— 运维必须知道"丢了上报"， +/// 而不是让 Redis 在无界增长里悄悄 OOM。 async fn forward_to_redis(redis: &redis::aio::ConnectionManager, msg: &DeviceMessage) { let json_str = match serde_json::to_string(msg) { Ok(json) => json, @@ -61,20 +84,40 @@ async fn forward_to_redis(redis: &redis::aio::ConnectionManager, msg: &DeviceMes }; let mut conn = redis.clone(); - let result: Result<(), _> = redis::cmd("LPUSH") + // 一条 pipeline 完成：入队 → 取裁剪前长度 → 裁剪到上限 → 取裁剪后长度 + // （4 条命令 1 次往返；`ignore()` 的两条不占返回位） + let mut pipe = redis::pipe(); + pipe.cmd("LPUSH").arg("device:reports").arg(&json_str).ignore(); + pipe.cmd("LLEN").arg("device:reports"); + pipe.cmd("LTRIM") .arg("device:reports") - .arg(&json_str) - .query_async(&mut conn) - .await; + .arg(0) + .arg(REPORTS_QUEUE_MAX - 1) + .ignore(); + pipe.cmd("LLEN").arg("device:reports"); + + let result: Result<(i64, i64), redis::RedisError> = pipe.query_async(&mut conn).await; match result { - Ok(_) => { - tracing::debug!( - "[forward] 已推送 dev_id={} act={}", - msg.dev_id.as_deref().unwrap_or("?"), - msg.act - ); + Ok((len_before_trim, len_after_trim)) => { + if len_before_trim > len_after_trim { + // 越界裁剪：明确告知丢弃了多少条（这些上报已不可恢复，需要人工关注消费端状态） + tracing::warn!( + "[forward] device:reports 超过上限 {}，已裁剪最旧的 {} 条上报（长度 {} → {}）", + REPORTS_QUEUE_MAX, + len_before_trim - len_after_trim, + len_before_trim, + len_after_trim + ); + } else { + tracing::debug!( + "[forward] 已推送 dev_id={} act={} queue_len={}", + msg.dev_id.as_deref().unwrap_or("?"), + msg.act, + len_after_trim + ); + } } - Err(e) => tracing::error!("[forward] Redis LPUSH 失败: {}", e), + Err(e) => tracing::error!("[forward] Redis LPUSH/LTRIM 失败: {}", e), } }