12 Commits

Author SHA1 Message Date
21eb52df24 chore: v0.1.4 — 版本号收拢(6 处)
internal/version/version.go(ldflags 示例 + 默认值)、scripts/release.sh(注释 + 默认值)、
scripts/build_installers.sh、packaging/rpm/webui4frpc.spec(Version + changelog,含本次
负载均衡四条)、README.md 与 packaging/windows/README.md 的安装示例。

漏一处就会让某个安装包的示例文件名指向不存在的产物,所以按 v0.1.3 的 5 处清单逐个核对,
并补上 084ef62 才纳入的 packaging/windows/README.md,共 6 个文件。
2026-10-06 16:38:00 +08:00
1a850b4846 docs: README 补负载均衡(自导式释放)与集群页开关
## 新增

- **特性 → 集群模式**:负载均衡条目,点明**没有中心调度**及其原因。
- **运行 → 负载均衡参数**:`W4F_BALANCE_MODE` / `W4F_BALANCE_COOLDOWN` /
  `W4F_BALANCE_INTERVAL`,并写明集群页的开关**优先于**环境变量(页面上选过就以
  页面为准,重启不丢)。
- **集群模式 → 新小节「负载均衡(自导式释放)」**:规则本体、四条边界
  (自限到平摊值 / 不低于下限 / 不碰已停用转发 / **释放 ≠ 停止**)、三种模式、
  只读报告的字段。
- **API 表**:`GET /cluster/balance`(read)、`POST /cluster/balance/mode`(write)。
- **架构 / 视图 / 测试**:补上负载均衡;并修正一处失效引用
  `npm run type-check` → `npm run build`(web/package.json 的 scripts 里根本没有
  type-check,照原文执行会报 missing script)。

## 为什么这两个「为什么」要写进文档

1. **为什么只能自导**:负载信号是持有的转发数,而这个数只在**自己的拓扑条目消失**
   时才下降,条目又是本地簿记 —— 所以 leader 只能*看见*失衡,无法*消除*它;要消除
   就得伸手改另一个节点的簿记,而这正是这个环到处都在避免的跨节点写。
2. **为什么「释放」不能复用「停止」**:停止保留条目并标记链路停用;均衡释放必须
   **删除条目**且保持链路启用。复用停止的话,每个节点的重复认领守卫都会拒绝这条
   已是「有主」的转发,它会被永久搁死在刚放弃它的那台机器上。

这两条是设计约束而不是实现细节:后续任何人改这块,不知道原因就会改错。

## 默认 dry-run 的理由也写明了

释放会停掉该转发的 frpc worker,其上的**在途连接会断一次**(新节点认领后恢复),
所以默认只报告不动作;模式开关走 write 级权限。

go test ./... 全绿(go clean -testcache 后)。
2026-10-06 15:59:28 +08:00
c90d4f3791 feat(cluster,webapi): 集群页可切换负载均衡模式(每节点一个开关)
## 为什么之前没有开关(以及为什么现在这么加)

模式原本只读一次环境变量 `W4F_BALANCE_MODE`,UI 只显示「观察模式 / 已启用」。
把开关加进来时有两个坑必须一起解决,否则开关是个摆设:

1. **不持久化就等于骗人**:只用 env,页面上翻一下、重启一次就回到默认,
   而页面已经显示成功了。
2. **不能做成集群级开关**:每个成员只释放自己的转发,根本没有一个动作能把它们
   一起翻过来。做成集群开关就是撒谎。

所以:**每节点一个开关**,写的是本节点自己的设置。

## 持久化与优先级

- 新增 `settings` 表的 `balance_mode` 列 + `Store.BalanceMode()`。**单独一列**,
  不是塞进 settings JSON:那个 JSON 是整份重写的,而这是会被各节点同时写、必须
  各自收敛的状态,跟 `nodeKey`/`settings` 同一类处理。
- **存了就以存的为准**,env 只作首次启动的默认值;`Settings()` 读回时带上它,
  但 `UpdateSettings` 只在非空时镜像到列,所以一次无关的设置保存不会把它清掉。
- `SetModeFromUI` **先落盘再应用**:写失败就什么都不改,否则节点会运行一个重启后
  复现不出来的模式——开关就成了谎言。
- 认不出的模式一律拒绝(400),不猜。

## 报告新增两个字段

- `modeSource`: `env` | `stored` —— 说清当前模式是启动参数给的还是有人在这里选的。
- `canControl`: 由**调用者的权限**决定,只读身份不会看到一个点了会 403 的开关。

`PUT /cluster/balance/mode` 走 **write** 级鉴权:释放会停 worker、断活连接。

## 两处我自己抓到并修掉的错

1. **`ModeSource()` 一开始写成「能存就算 stored」**(判据是 `persist != nil`),
   于是**每个有 store 的节点都自称 stored**,哪怕从没人设过。canary 第一轮就露了:
   `mode=dry-run source=stored`。改成看「是否真的存过值」。
2. **启动日志打的是 env 变量而不是生效值**,存过 `on` 之后仍打 `DRY-RUN`——
   上一行的 `loaded from settings` 还在,两行自相矛盾,正好会让人以为释放是关的
   而它其实开着。

## 测试

新增 6 条:存的模式压过 env、没存就回落 env、落盘失败则模式不变、乱值被拒、
UI 切换同时落盘并生效、无持久化能力时不宣称可控制。
canary 实跑 6 步:初始 `dry-run/env` → 设 `on` → 复查 `on/stored` → 乱值 400 →
乱值后模式不变 → **重启后仍 `on/stored`**,启动日志 `load balancing ON (source=stored …)`。
vue-tsc 干净;go build / vet / go test ./... 全绿。
2026-10-06 14:37:18 +08:00
790eaa5ed1 feat(web): 集群页负载均衡改为「本机份额」视图
跟着 c69fb2c 的模型改。旧 UI 是中心调度视角(超载阈值 = Score × 均值、
leader 会迁 X→Y),新模型里每台节点只管自己的转发,所以页面必须**以本机为主语**。

## 变化

- 标题徽标从「观察模式 / 已启用」+「本节点非 leader」改为
  **「观察模式(不释放)/ 已启用(会释放)」+「本机超出份额 / 本机在份额内 / 非集群成员」**
  —— 没有 leader 这回事了,每台都在判断。
- 阈值行换成真正的规则:
  `本机转发 N 条 > 平摊份额 M 且至少 2 条时,本机自行释放一条`,
  并给出 `集群共 X 条 / Y 台存活` 与冷却间隔。
- 节点条改成**按转发数排序**(最忙的在前),条宽按最大转发数归一化;
  「超载」徽标改为「超出份额」(判据是转发数 vs 份额,不再看 Score)。
  Score 从卡片上去掉——新规则不用它,留着会让人以为它参与了判断。
- 计划区按 blockedBy 分四种:释放进行中 / 将释放 / 冷却中(含「已过 N 分钟」)/
  暂不释放(把机器原因翻成人话)/ 单节点 / 已均衡(给出具体条数与份额)。
- 加一句说明:每台各自判断、自己撤销一条丢回环、由最闲者认领、无中心调度。

## 一处刻意的设计

`blockedBy` 的文案单独做映射,因为「超出份额但不动」正是最容易被误读成故障的
状态——渲染成人话(冷却期未过 / 上一次释放尚未确认 / 可释放的转发都已停用)
才是这个字段存在的意义。

vue-tsc --noEmit 干净;npm run build 通过(中途踩到 Vue 模板重复 class 属性
导致解析失败,已修);canary 实跑端点 200、字段齐全、默认 dry-run;
go build / go test ./... 全绿。
2026-10-06 14:06:23 +08:00
c69fb2caba feat(cluster): 负载均衡改为每节点自主「先撤销本地转发,再丢回环」
按用户要求重做:不再由 leader 单独调度,而是**每个节点对自己的转发跑同一条
规则**,超了就自己撤销一条、把任务丢回环上,由环原有的「最闲者认领」决定它
落到哪台。

## 为什么必须是 owner 自己动手(不是洁癖,是实测逼出来的)

真机探针(跑完已删)显示,真实 revoke 之后:

    TopologyOwner after revoke = "n1"     ← 条目还在(只标记不删除)
    n1 Forwards=3  score=310.0            ← 负载根本没降
    after step: revoking=true publishing=false

⇒ 一节点的负载**只有它自己的拓扑条目消失时才会下降**,而条目是本地簿记,没有任何
对端可以代它写。leader 从外面只能看见失衡、无法让失衡成为事实——它必须伸手去改
另一个节点的簿记,而这正是这个环到处都在避免的跨节点写(revoke / restart 定向往
owner 也是同一个理由)。

⇒ 而且旧实现**根本跑不起来**:phase 1 判据是「条目消失」,可 mark-don't-remove
让它永远不消失,于是每 120s 打一条 `revoking … timed out; abandoning`,
**phase 2 永不执行,一条转发都不会迁**。开 `W4F_BALANCE_MODE=on` 只会刷日志。

## 规则(替代原来的 ratio 阈值)

    当 Forwards >= 2 且 Forwards > ceil(总数/存活节点数) 时,撤销一条

- **自限**:3 条 / 3 台 → owner 2 > 1,卸一条 → 1,`1 > 1` 不成 → 停。
  没有上界的话 owner 会把转发**全部卸光**。
- **不会卸空**:低于下限(2)永不撤销。
- 平摊值向上取整(3/3=1,不是 0),否则 1 条 / 3 台的节点也算「超了」。
- **不再用 mem/net 参与判据**:实测 `.30` 空闲时 netPct 能到 55%,Score 会高于
  有转发的节点——用 Score 会把纯网络抖动当成负载。
- 节奏用**每节点冷却**(默认 600s)而不是「失衡持续多久」窗口:规则本身已经要求
  真实失衡,没有需要平滑的瞬态,只有需要限制的抖动频率。

## 为什么必须是新的 Release 任务类型,不能复用 Revoke

用户停止(Revoke)**保留拓扑条目**作为 disabled 标志的载体;均衡释放必须**删除
条目**,因为 (a) 负载由自己的条目数算出,条目不删负载不降;(b) 每个节点的
duplicate-guard 会拒绝「已有 owner 的转发」的创建任务——保留条目会让**所有节点
都拒领**,转发被永久搁死在刚放弃它的那台机器上。

Release 因此:`Handler.Release`(停 worker,但**不标记 stopped**)→ 删条目 →
`SubmitTask` 以**启用状态**重新发布。它需要自己的 Handler 方法,不能回落到
Revoke:Revoke 会顺手把 link 标成 stopped,而这条转发即将在别的节点上跑。

## 顺带修掉两个既有缺口(实测确认)

1. **`submitMigrate` 发布不出去**:`OnToken` 的 localPending 重合并只保留「从未
   publish 过」的本地 pending,而只有 `Add*` 系列会记 `published`。引擎里直接写
   `PendingTasks` 的任务会在下一次 adopt 时被静默丢弃。
   ⇒ 新 `ReleaseTask` 走 `AddPending`。
   (`SubmitRestart` 有同样的毛病,但它不在本次范围,未改动——见提交说明末尾。)
2. `Release` 需要绕过 duplicate-claim 守卫(同 Restart),否则「owner 本来就持有
   这条转发」这一前提会让每条 Release 都被丢掉。

## 测试(17 条,全绿)

规则:平摊值取整、超了才撤、撤到平摊就停、单节点不撤、低于下限不撤、不撤已停
转发、候选选择确定。
机制:Revoke 留载体 vs Release 删条目(这是两种语义的核心差异)、Release 只由
owner 执行且不消费他人的任务、owner 释放后环的普通认领把它接走。
模式:dry-run 不动任何东西、off 真的什么都不做、未知模式回落 dry-run、
冷却限制抖动频率、确认超时清理。
只读:Report 不变更任何状态、能解释「超了但不动」的原因、均衡节点给出真实答案。

**双向验证 3 条**(改坏→变红→还原→变绿):
- 去掉 Release 的 duplicate-guard 例外 → `the owner released 0 times, want 1`
- 去掉「不超过平摊值」上界 → `2 owned against share 2 must not release`
- Release 改成只标记不删除 → `n1 should now own 2, got 3`

go build / go vet / go test ./... 全绿(go clean -testcache 后逐包实跑)。

## 未做 / 待办

- `SubmitRestart` 的同类发布缺口未修(同样会被 adopt 丢弃),需要单独确认它
  在实机上是否真的丢过 restart。
- UI 尚未跟进新报告结构(下一步)。
2026-10-06 14:03:03 +08:00
b40b53b40b feat(web): 集群页显示负载均衡视图
接 8176478 的只读端点。此前 balancer 的判断**只能从日志看**,而且只在"确实失衡
且迟滞窗口已满"时才输出 —— 运维在健康状态下得到的是沉默,与"负载均衡没开"完全
无法区分。

## 界面

- **各节点 Score 条**:按分数降序(最热的在顶部),条宽按最大分数归一化 ——
  空闲时(score≈5)和繁忙时(score≈500)都能看出形状。
- **阈值说明**:直接写出"超载 = Score > 均值 × N"与"需持续 N 分钟",运维不必
  去猜这两个数字怎么定的。
- **四种状态**(仅 leader 显示):
  - 已均衡 —— 无超载节点
  - 等待确认 —— 已超载但迟滞窗口未满,并显示"已持续 N 分钟 / 需 N 分钟"
    ★ 这是最有价值的一态:否则"为什么还不迁"无从回答
  - 将迁移 / 迁移进行中 —— 给出具体转发与两端节点
- **follower 不显示计划**,与端点的 isLeader 门禁一致:随便查一台节点不该
  暗示集群不会做出的决定。
- **端点 404 时整块不渲染** —— 缺席比空面板诚实(空面板会让人以为支持)。

## 一处刻意的非对称

`load()` 里 ring 失败保留上次值,balance 失败则**清空**。理由:ring 是拓扑快照,
陈旧可接受;balance 是**决策面**,端点开始失败后仍显示"已均衡"比什么都不显示
更危险 —— 会让人以为系统确认过集群是好的。

vue-tsc --noEmit 干净;go build / go test ./... (go clean -testcache) 全绿。
2026-10-06 12:58:17 +08:00
d347d2a648 fix(cluster): 把 balancer 真的接进 Handler(端点此前 404)
上一提交注册了 /cluster/balance 并加了 Handler.Balancer 字段,但构造 Handler 时
忘了赋值 —— 端点因 h.Balancer == nil 走 404 分支。

单测发现不了:它们直接构造 Balancer 调 Report(),从不经过 HTTP 层。
canary 实跑抓到:端点 404(无认证倒是正确 401)。修好后同一 canary 返回 200,
字段齐全、默认 dryRun=true、单节点自认 leader、wouldMigrate 为空。

教训:单元测试覆盖不到"接线",凡新增端点必须真的起进程打过一次。
2026-10-06 12:45:28 +08:00
8176478269 feat(cluster,httpapi): GET /cluster/balance 只读负载视图
## 为什么需要它

balancer 的判断原本只能从日志看,而且**只在"确实失衡 且 迟滞窗口已满"时才输出**。
运维问"这个集群均衡吗"——在健康状态下得到的是**沉默**,与"负载均衡没开"
完全无法区分。这个端点把它的判断做成可主动查询的。

## 关键性质:Report() 不扰动迟滞计时器

Report 只读,**不碰 overSince**。否则好奇的运维反复轮询这个端点就会不断重置
"失衡已持续多久"的计时,导致真正的迁移永远不会触发——一个"看一看就把它看没了"
的只读接口。测试 TestReportDoesNotDisturbHysteresis 正是钉住这点:轮询 5 次后
计时器必须原封不动,且随后满足窗口时迁移仍要发生。

## 只 leader 报计划

`wouldMigrate`/`overloadedSince` 仅在 `IsLeader()` 时给出。非 leader 仍返回各节点
Score(那是信息性的),但不报"会迁什么"——否则运维随便查一台节点会拿到与集群
实际行为不符的答案。TestReportOnFollowerHasNoPlan 钉住。

## 返回内容

    nodes[]           各节点 Forwards/MemPct/NetPct/Score
    isLeader          只有 leader 会行动
    overloadedSince   超载已持续多久(迟滞进度)
    wouldMigrate      窗口满足后会迁哪条、从哪到哪
    inFlight          两阶段迁移进行中(revoke 慢 vs 卡死可区分)
    dryRun / enabled  当前是观察还是真迁

## 接线

Balancer 在 NewBalancer 时就创建(HTTP 层需要),W4F_BALANCE_* 解析完再用
SetRatioAndSustain 应用;新增 SetDryRun / DryRun / SetRatioAndSustain,
避免初始化顺序耦合。Handler.Balancer 为 nil 时端点返回 404。

## 测试

新增 4 条:Report 不扰动迟滞(轮询后计时器不变、且迁移仍能触发)、
follower 无计划、窗口满足前后计划的有无、in-flight 可见。
go clean -testcache 后逐包实跑 7/7 全绿。
2026-10-06 12:44:17 +08:00
45a6a76621 feat(cluster): 负载均衡接入运行时 —— 默认 dry-run,显式 opt-in 才真正迁移
承接 7d5bca5(Balancer 状态机)。本提交接线到 main.go,并让 **dry-run 成为默认**。

## 为什么默认不启用

线上此刻是 `.30` 持 3 条转发、另两台 0 条。按已实现的规则,这会在第一次
检查就立刻触发迁移 —— 而每条迁移都要停一个 frpc worker、断掉活跃连接。
在没有看过真实数据会迁什么之前就把这个开关打开,等于拿生产连接赌一把。

所以:

    W4F_BALANCE_MODE=dry-run   只报告不动手(默认)
    W4F_BALANCE_MODE=on        真正迁移
    W4F_BALANCE_RATIO=2        超载 = Score 超过均值 2 倍
    W4F_BALANCE_SUSTAIN=1800   失衡需持续 30 分钟
    W4F_BALANCE_INTERVAL=10    每 10 秒检查一次

dry-run 走同一套决策与迟滞逻辑(stepDryRun 复用 planBalance),因此它报告的
就是真开会迁的东西 —— 先用它观察,确认符合预期再切 on。

## 只有 leader 平衡

WatchBalance 里 `if !b.engine.IsLeader() { continue }`。搬转发是全集群决策,
令牌是共享的:若每个成员各自决定,必然出现两个节点同时对同一条转发发 revoke。
(这与 Restart/RemoveNode 走同样的 owner 定向是同一类约束。)

## envFloat

新增:环境变量解析失败时回落到默认值并打日志,而不是静默变成 0(0 会被
NewBalancer 当成非法值钳制,但静默钳制无法与"用户显式要 0"区分)。

## 测试

- TestDryRunNeverTouchesTheRing —— dry-run 必须报告"would migrate"且**不发布任何
  任务、不动拓扑、不进入 in-flight**。这是它作为默认值存在的全部意义。
- TestOnlyLeaderBalances —— 非 leader 跑 WatchBalance 不得发布任何任务。

go build / go vet / go test ./...(go clean -testcache 后)全绿,gofmt 干净。

## 下一步(未包含)

集群页 UI 展示超载提示与迁移进度;以及**先以 dry-run 在生产跑一轮观察**,
确认 planBalance 对真实数据的判断符合预期后再决定是否切 on。
2026-10-06 11:56:23 +08:00
7d5bca5b72 feat(cluster): 负载均衡迁移 —— 定向投递 + 两阶段提交
承接 875e580(只落决策逻辑)。本提交补齐运行时:周期检查、迟滞判定、
两阶段迁移,以及让迁移任务**定向**到指定节点。

## 为什么必须是定向的(核心安全点)

loadSnapshot() 的 Forwards 来自 ForwardsOwnedBy,所以旧 owner 一旦被摘掉条目,
它立刻变成最闲节点。而普通创建任务是"任何最闲者认领"—— 于是它会把刚迁走的
转发**当场抢回**,迁移变成"只重启一次 worker"的空操作。

故 Task 新增 Migrate + Target 字段,runCommands 的选择循环照 Restart 的范式
处理:Target == 自己才认领,否则 `continue` 让任务随环前进。

## 为什么必须两阶段(不能一次发 revoke + submit)

1. 旧 owner 变最闲后会把未定向的任务抢回(上面那条);
2. PendingList 按 `t<ID>` **字符串**排序(`t10 < t5`),同轮内两个任务的先后
   不可控。

Balancer 因此是显式状态机:phase1 发布 revoke 并等待旧条目真正消失,确认后才
发布 phase2 的定向 migrate。两个超时兜底:revoke 超时则放弃;migrate 超时
(target 没认领)则退回**无定向** SubmitTask,让"任何最闲者"接手 —— 转发不会
被丢,只是延迟。这与环本身的行为一致(实测:只要还有更闲的节点,pending 任务
最终会被认领;只有"全员都更忙"时才滞留)。

## 两个被测试逼出来的设计修正

- **绝对下限 minForwards=2**:只有"超均值倍数"在小集群失真——3 节点 1 条转发时
  owner Score≈105 而两台空闲≈5,20 倍于均值会触发迁移,但那是完全正常的分布,
  迁它只会断一条活跃连接。倍数负责看形状,下限负责"有没有东西值得搬"。
- **Migrate 也要绕过 duplicate-claim 守卫**,否则旧条目还在时目标节点会拒绝
  认领(这条守卫正是为了防止同一 remotePort 注册两次 → frps 报
  "proxy already exists")。

## 测试(本项目第六次假绿的教训)

新增 6 条,全绿。做双向验证时发现
TestMigrateTaskIsAddressedNotClaimedByLightest **删掉定向分支后仍然 PASS**,
查因:`loadSnapshot()每次 OnToken 都用 Handler.RuntimeLoad() 刷新 mem/net,
我手工设的 State.Nodes[].Load 被静默覆盖 ⇒ n2/n3 始终平手 ⇒ "最闲者"恰好
就是 target,定向性从未被验证。

改为**用三个 Handler 的不同 RuntimeLoad 制造真实差异**(n3 最闲、n2 是较重的
target)后,删掉定向分支如期变红:

    n3 (the LIGHTEST node) claimed 1 migrate task(s); the task is addressed
    to n2 and must ignore load entirely

⇒ 这条纪律再次生效:测试必须制造出能让错误实现暴露的**不对称**,否则"恰好通过"
比"失败"更危险。

go build / go vet / go test ./... 全绿,gofmt 干净。
2026-10-06 11:50:53 +08:00
875e5805eb feat(cluster): 负载均衡迁移的决策逻辑(planBalance)
用户提出「加负载均衡算法,定期重新分配任务」。本提交只落**决策**部分:
给定当前环快照,算出「该不该迁、迁哪条、迁给谁」。运行时接线未包含在内。

## 决策规则(刻意保守)

- **阈值**:某节点 Score 超过均值 minOverRatio 倍,且与最闲节点相差**至少一条转发的量**
  (Score 里 Forwards 权重 100,所以差 <100 视为噪声)——避免为 1% 的差异重启 worker
- **一次一条**:负载抖动时不会引发一批 worker 同时重启(每次迁移都断活跃连接)
- **不动停用的转发**:迁走 disabled 的转发等于把用户明确关掉的东西重新拉起来
- **不动 offline 节点**:否则会迁到一台不存在的机器上
- **完全均衡时返回"无事可做"**:这就是阈值触发的意义——健康环零开销

## 尚未实现(本提交不含)

1. **持续时间判定**:阈值命中需持续 N 分钟才动手,避免瞬时尖峰触发迁移。
   当前 planBalance 是无状态的,时间维度留给调用方的 BalanceTracker。
2. **两阶段提交**:这是**安全性关键**,不能简单地把 RevokeTask 与 SubmitTask
   一起发出去。实测 PendingList 按 `t<ID>` 字符串排序(`t10 < t5`),同轮内
   顺序不可控;且旧 owner 停掉 worker 后自己的 Load 立刻下降,很可能又成为
   最闲节点而把刚迁走的转发重新认领回去。正确做法是先确认旧条目已消失,
   再提交新任务。
3. **周期性检查与 API/UI 暴露**。

## 测试

4 条,均针对 planBalance:搬走超载节点的一条转发 / 不搬停用转发 / 均衡时不动 /
不选 offline 节点。构造 seed 时经 JSON 往返让每个 State 持有独立 Nodes 切片
(AdoptState 是值赋值但 Nodes 是切片,共享底层数组会造成假绿——这个坑本项目
已踩过多次)。
2026-10-06 11:14:25 +08:00
084ef62b32 docs: 同步 packaging/windows/README.md 的安装示例到 0.1.3
由 scripts/release.sh 在发版时自动改写(deb/rpm/setup.exe 文件名带版本号)。
主 README.md 已在 v0.1.3 的版本号提交里同步,此处补齐打包目录下的副本。
2026-10-06 11:11:36 +08:00
18 changed files with 2210 additions and 246 deletions

View File

@ -32,6 +32,14 @@
- 主负载信号:当前节点拥有的转发数(权重 ×100) - 主负载信号:当前节点拥有的转发数(权重 ×100)
- 次级信号:**NIC 网络接口利用率**(从 `/proc/net/dev` 增量 + `/sys/class/net/*/speed` 计算,排除了 lo/docker/veth 等虚拟接口) - 次级信号:**NIC 网络接口利用率**(从 `/proc/net/dev` 增量 + `/sys/class/net/*/speed` 计算,排除了 lo/docker/veth 等虚拟接口)
- 同转发数节点间由内存占用 + 网络饱和度打破平局 - 同转发数节点间由内存占用 + 网络饱和度打破平局
- **负载均衡(自导式)**:每个节点对**自己的**转发跑同一条规则 —— 持有数超过
`ceil(总转发数 / 存活节点数)` 且至少 2 条时,自行撤销一条、把任务丢回环上,
再由环按「最闲者优先」重新认领
- **没有中心调度**:一节点的负载只有它自己的拓扑条目消失时才会下降,而条目是
本地簿记,没有对端能代它写,所以只能由持有者自己动手
- 规则自限:撤到平摊值即停(否则会把转发卸光),且低于下限永不撤、不会卸空节点
- 每节点默认 10 分钟冷却,限制抖动频率;默认 **dry-run**(只报告不释放)
- 集群页每节点一个开关:关闭 / 观察模式 / 启用,可持久化
- **per-forward 独立进程**:每条转发拥有独立的 frpc 配置文件和 frpc 进程(worker 键 = `local~remote~port`),单条转发故障不影响兄弟,消除旧模型的多节点 `proxy already exists` 冲突 - **per-forward 独立进程**:每条转发拥有独立的 frpc 配置文件和 frpc 进程(worker 键 = `local~remote~port`),单条转发故障不影响兄弟,消除旧模型的多节点 `proxy already exists` 冲突
- **容错自愈**:leader / 非 leader 节点宕机 → 自动检测 → 环跳过死亡节点、任务重挂;leader 宕机 → 上邻居晋升为新 leader - **容错自愈**:leader / 非 leader 节点宕机 → 自动检测 → 环跳过死亡节点、任务重挂;leader 宕机 → 上邻居晋升为新 leader
- **宕机自动重联**:节点宕机重启后通过缓存的集群拓扑(对端地址 + 准入密钥)自动重联,显式脱离集群的节点不自动重联 - **宕机自动重联**:节点宕机重启后通过缓存的集群拓扑(对端地址 + 准入密钥)自动重联,显式脱离集群的节点不自动重联
@ -79,7 +87,7 @@ cd web && npm run build && rm -rf internal/httpapi/dist && cp -r web/dist intern
### Linux (Debian / Ubuntu) ### Linux (Debian / Ubuntu)
```bash ```bash
sudo dpkg -i webui4frpc-0.1.3-linux-amd64.deb # 或 linux-arm64.deb sudo dpkg -i webui4frpc-0.1.4-linux-amd64.deb # 或 linux-arm64.deb
sudo nano /etc/webui4frpc/webui4frpc.env # 改掉默认密码 sudo nano /etc/webui4frpc/webui4frpc.env # 改掉默认密码
sudo systemctl restart webui4frpc sudo systemctl restart webui4frpc
``` ```
@ -87,7 +95,7 @@ sudo systemctl restart webui4frpc
### Linux (RHEL / Fedora / openEuler / 默认 / 龙蜥) ### Linux (RHEL / Fedora / openEuler / 默认 / 龙蜥)
```bash ```bash
sudo rpm -i webui4frpc-0.1.3-linux-x86_64.rpm sudo rpm -i webui4frpc-0.1.4-linux-x86_64.rpm
sudo vi /etc/webui4frpc/webui4frpc.env sudo vi /etc/webui4frpc/webui4frpc.env
sudo systemctl enable --now webui4frpc sudo systemctl enable --now webui4frpc
``` ```
@ -97,7 +105,7 @@ sudo systemctl enable --now webui4frpc
### Windows ### Windows
双击 `webui4frpc-0.1.3-windows-amd64-setup.exe`(ARM 设备用 `windows-arm64-setup.exe`)。 双击 `webui4frpc-0.1.4-windows-amd64-setup.exe`(ARM 设备用 `windows-arm64-setup.exe`)。
安装向导会询问监听地址与管理员凭据,完成后: 安装向导会询问监听地址与管理员凭据,完成后:
- 程序装到 `C:\Program Files\webui4frpc` - 程序装到 `C:\Program Files\webui4frpc`
@ -107,8 +115,8 @@ sudo systemctl enable --now webui4frpc
### macOS ### macOS
```bash ```bash
sudo installer -pkg webui4frpc-0.1.3-darwin-arm64.pkg -target / # Apple Silicon sudo installer -pkg webui4frpc-0.1.4-darwin-arm64.pkg -target / # Apple Silicon
sudo installer -pkg webui4frpc-0.1.3-darwin-amd64.pkg -target / # Intel sudo installer -pkg webui4frpc-0.1.4-darwin-amd64.pkg -target / # Intel
``` ```
或直接双击 `.pkg` 走图形安装器。安装后 launchd 开机自启,配置在 或直接双击 `.pkg` 走图形安装器。安装后 launchd 开机自启,配置在
@ -126,7 +134,7 @@ sudo launchctl kickstart -k system/com.jianfeeeee.webui4frpc
不想装服务只想跑一下,用对应的 `.tar.gz` / `.zip`,解压即用: 不想装服务只想跑一下,用对应的 `.tar.gz` / `.zip`,解压即用:
```bash ```bash
tar xzf webui4frpc-0.1.3-linux-amd64.tar.gz tar xzf webui4frpc-0.1.4-linux-amd64.tar.gz
./webui4frpc -addr 127.0.0.1:7500 -user admin -password 你的密码 -workdir ./data ./webui4frpc -addr 127.0.0.1:7500 -user admin -password 你的密码 -workdir ./data
``` ```
@ -160,6 +168,19 @@ sha256sum -c SHA256SUMS --ignore-missing
环境变量 `W4F_HOST` 可覆盖节点对外可达的主机名(Docker DNS / 通配监听场景用)。 环境变量 `W4F_HOST` 可覆盖节点对外可达的主机名(Docker DNS / 通配监听场景用)。
### 负载均衡参数
每节点独立生效;集群页的开关**优先于**这里的默认值(在页面上选过就以页面为准,重启不丢)。
| 环境变量 | 默认 | 说明 |
| --- | --- | --- |
| `W4F_BALANCE_MODE` | `dry-run` | `off` 不参与 / `dry-run` 只报告 / `on` 真正释放 |
| `W4F_BALANCE_COOLDOWN` | `600` | 同一节点两次释放之间的最小秒数 |
| `W4F_BALANCE_INTERVAL` | `10` | 检查间隔(秒) |
**释放会停掉该转发的 frpc worker,其上的在途连接会断一次**,由新节点认领后恢复;
因此默认是 `dry-run`。集群页的开关走 write 级权限。
### 启动行为 ### 启动行为
节点启动时的集群引导优先级: 节点启动时的集群引导优先级:
@ -191,10 +212,10 @@ webui4frpc(单二进制,无 frp 依赖,前端 go:embed)
│ ├── render 渲染 frpc JSON 配置(per-forward 单 proxy + 高级参数 + LB/健康检查) │ ├── render 渲染 frpc JSON 配置(per-forward 单 proxy + 高级参数 + LB/健康检查)
│ ├── process worker 进程管理(spawn/stop/restart/自愈/日志轮转,worker 键 = local~remote~port) │ ├── process worker 进程管理(spawn/stop/restart/自愈/日志轮转,worker 键 = local~remote~port)
│ ├── install 一键下载官方 frpc(GitHub Releases,含解压安全防护) │ ├── install 一键下载官方 frpc(GitHub Releases,含解压安全防护)
│ ├── cluster 令牌环协议(ring engine / leader 选举 / 容错 / 宕机重联 / 二进制分发 / NIC 负载采样) │ ├── cluster 令牌环协议(ring engine / leader 选举 / 容错 / 宕机重联 / 二进制分发 / NIC 负载采样 / 自导式负载均衡)
│ └── httpapi REST API + 静态资源(auth/session / canvas / forwards / cluster / users / audit 导出) │ └── httpapi REST API + 静态资源(auth/session / canvas / forwards / cluster / users / audit 导出)
└── web/ Vue 3 SPA(Element Plus + VueFlow 画布 + Pinia) └── web/ Vue 3 SPA(Element Plus + VueFlow 画布 + Pinia)
└── src/views/ 状态总览 / 连接配置(画布)/ 集群 / 设置 / 账号与密钥 └── src/views/ 状态总览 / 连接配置(画布)/ 集群(含负载均衡开关)/ 设置 / 账号与密钥
``` ```
## API ## API
@ -220,6 +241,8 @@ webui4frpc(单二进制,无 frp 依赖,前端 go:embed)
| | GET | `/binary/status` | read | frpc 路径 | | | GET | `/binary/status` | read | frpc 路径 |
| | POST | `/binary/install` | write | 安装 frpc | | | POST | `/binary/install` | write | 安装 frpc |
| **集群** | GET | `/cluster/ring` | read | 令牌环快照 | | **集群** | GET | `/cluster/ring` | read | 令牌环快照 |
| | GET | `/cluster/balance` | read | 本节点负载均衡视图(只读,不变更状态) |
| | POST | `/cluster/balance/mode` | write | 切换本节点负载均衡模式(持久化) |
| | POST | `/cluster/create` | write | 创建独立集群 | | | POST | `/cluster/create` | write | 创建独立集群 |
| | POST | `/cluster/join-ring` | write | 加入集群(本节点发起) | | | POST | `/cluster/join-ring` | write | 加入集群(本节点发起) |
| | POST | `/cluster/join` | write | 接收新节点加入 | | | POST | `/cluster/join` | write | 接收新节点加入 |
@ -254,8 +277,45 @@ webui4frpc(单二进制,无 frp 依赖,前端 go:embed)
- **任务分发**:在任意节点画布上配置 `localOnly=false` 的转发并保存 → 画布差异判断生成命令 → 令牌携带 → 负载最低节点摘取 → 拉起 frpc worker - **任务分发**:在任意节点画布上配置 `localOnly=false` 的转发并保存 → 画布差异判断生成命令 → 令牌携带 → 负载最低节点摘取 → 拉起 frpc worker
- **容错**:节点宕机 → 自动检测 → 环自愈(跳过死亡节点 / leader failover);重启 → 自动重联 - **容错**:节点宕机 → 自动检测 → 环自愈(跳过死亡节点 / leader failover);重启 → 自动重联
- **退出集群**:在集群页「退出集群」→ 本节点脱离为 standalone,不自动重联 - **退出集群**:在集群页「退出集群」→ 本节点脱离为 standalone,不自动重联
- **负载均衡**:见下节
集群页展示:环拓扑(self/leader 标记)、令牌轮次、节点负载、待办任务、活跃转发拓扑(含 owner 归属)、增量事件日志。 集群页展示:环拓扑(self/leader 标记)、令牌轮次、节点负载、待办任务、活跃转发拓扑(含 owner 归属)、负载均衡(本机份额与释放计划)、增量事件日志。
### 负载均衡(自导式释放)
**为什么不是 leader 统一调度**:一节点的负载信号是它持有的转发数,而这个数只在
**它自己的拓扑条目消失**时才下降 —— 条目是本地簿记。所以 leader 从外面只能*看见*
失衡,无法*消除*它;要消除就得伸手改另一个节点的簿记,而这正是这个环到处都在避免
的跨节点写(撤销、重启同样都是定向往 owner)。
于是规则由每个节点对自己的转发本地执行:
```
当 持有转发数 >= 2 且 持有转发数 > ceil(总转发数 / 存活节点数)
则 撤销其中一条 → 删除条目 → 以启用状态重新发布到环 → 由最闲节点认领
```
- **自限**:撤到平摊值即停。3 条分 3 台时,持有 2 条的那台卸一条后变成 1 条,
`1 > 1` 不成立 → 停手。没有这个上界,owner 会把转发全部卸光。
- **不会卸空**:低于下限 2 条永不撤销。
- **不碰已停用的转发**:停用条目的 entry 是用户「停用」决定的载体,释放它等于把
一条**已停**的转发当活干交给别人。
- **释放 ≠ 停止**:用户停止保留条目并标记链路为停用;均衡释放**删除条目**且链路保持
启用。两者不能混用 —— 复用停止的话,每个节点的重复认领守卫都会拒绝这条已是「有主」
的转发,它会被永久搁死在刚放弃它的那台机器上。
三种模式(集群页每节点一个开关,也可用 `W4F_BALANCE_MODE` 设默认值):
| 模式 | 行为 |
| --- | --- |
| `off` | 不参与均衡 |
| `dry-run` | 只报告「将释放哪条」,不动任何东西(默认) |
| `on` | 真正释放 |
只读接口 `GET /cluster/balance` 返回本节点视角:`selfForwards` / `share` /
`totalForwards` / `aliveNodes` / `selfOverShare`、`wouldRelease`、`releasing`、
`blockedBy`(为何暂不动的机器原因)、`mode` / `modeSource` / `canControl`。
该接口**不改变任何状态** —— 反复轮询不会启动、延迟或取消一次释放。
## 测试 ## 测试
@ -263,6 +323,9 @@ webui4frpc(单二进制,无 frp 依赖,前端 go:embed)
# 后端 # 后端
go test ./... go test ./...
# 单包(负载均衡规则与状态机)
go test ./internal/cluster/ -run TestShare -v
# 前端类型检查 # 前端类型检查
cd web && npm run type-check cd web && npm run build
``` ```

View File

@ -15,6 +15,7 @@ import (
"os" "os"
"os/signal" "os/signal"
"path/filepath" "path/filepath"
"strconv"
"strings" "strings"
"syscall" "syscall"
"time" "time"
@ -289,6 +290,21 @@ func main() {
log.Printf("ring[%s] revoked task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) log.Printf("ring[%s] revoked task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
return nil return nil
}, },
// ReleaseFn stops this forward's worker for a REBALANCE, leaving the
// link ENABLED. This must not reuse RevokeFn: that marks the link
// stopped, and the released forward is about to run on another node —
// a "stopped" flag would make the claim path treat the re-published
// task as a disabled forward and refuse to start a worker for it,
// leaving the forward dead everywhere.
ReleaseFn: func(ctx context.Context, tk *cluster.Task) error {
key := process.WorkerKey(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
if _, running := pm.Status(key); running {
_ = pm.Stop(key)
}
log.Printf("ring[%s] released worker %s (rebalancing): %s→%s:%d", selfID, key,
tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
return nil
},
// RestartFn brings an already-claimed forward back after a stop. It // RestartFn brings an already-claimed forward back after a stop. It
// must NOT run the claim bookkeeping (ClaimFn), which appends links and // must NOT run the claim bookkeeping (ClaimFn), which appends links and
// writes topology state: the forward is already established, and a stop // writes topology state: the forward is already established, and a stop
@ -388,6 +404,11 @@ func main() {
} }
}) })
// Load balancer is created here (before the Handler) so the HTTP layer can
// expose its read-only report. Mode/parameters are applied further down, once
// the W4F_BALANCE_* environment has been parsed.
balancer := cluster.NewBalancer(ring, 600)
h := &httpapi.Handler{ h := &httpapi.Handler{
Store: st, Store: st,
Process: pm, Process: pm,
@ -396,6 +417,9 @@ func main() {
Sessions: httpapi.NewSessionStore(), Sessions: httpapi.NewSessionStore(),
Cluster: reg, Cluster: reg,
Ring: ring, Ring: ring,
// Balancer is created above, before the environment that configures it is
// parsed, because the HTTP layer needs the pointer.
Balancer: balancer,
// SelfAddr must be reachable by peers. 0.0.0.0/empty on a real host // SelfAddr must be reachable by peers. 0.0.0.0/empty on a real host
// would break peer reconnect, so fall back to the hostname (container // would break peer reconnect, so fall back to the hostname (container
// name inside docker is a routable addr on the compose network). // name inside docker is a routable addr on the compose network).
@ -553,6 +577,65 @@ func main() {
go ring.WatchTokenLoss(clusterCtx) go ring.WatchTokenLoss(clusterCtx)
go ring.WatchLeader(clusterCtx) go ring.WatchLeader(clusterCtx)
// Load balancing — SELF-DIRECTED, and dry-run by default.
//
// Every node runs this check against its OWN forwards. There is no
// coordinator: a node's load only actually falls when its own topology entry
// goes away, so only the owner can produce the state change that makes the
// ring hand a forward to somebody else. The node gives a forward up and
// re-publishes it as an ordinary unclaimed task; the ring's normal
// lowest-load claim then places it. (The earlier design had the leader pick a
// victim and a destination — but it can only see the imbalance, not resolve
// it, so it had to reach into another node's bookkeeping to make its decision
// true.)
//
// A release stops a worker and drops a live connection, so acting on the ring
// is opt-in:
//
// W4F_BALANCE_MODE=dry-run report the decision, release nothing (default)
// W4F_BALANCE_MODE=on actually release
// W4F_BALANCE_MODE=off do nothing at all
// W4F_BALANCE_COOLDOWN=600 min seconds between two releases by one node
// W4F_BALANCE_INTERVAL=10 check every N seconds
balMode := strings.ToLower(strings.TrimSpace(os.Getenv("W4F_BALANCE_MODE")))
if balMode == "" {
balMode = cluster.BalanceDryRun
}
balInterval := time.Duration(envFloat("W4F_BALANCE_INTERVAL", 10)) * time.Second
if balInterval < time.Second {
balInterval = 10 * time.Second
}
balCooldown := int64(envFloat("W4F_BALANCE_COOLDOWN", 600))
balancer.SetCooldown(balCooldown)
balancer.SetMode(balMode)
// A mode stored from the cluster page wins over the environment, so flipping
// the switch is not silently reverted by the next restart. SetMode is applied
// first so a stored value only has to be a no-op when the two agree.
balancer.SetModePersistence(
func() (string, bool) {
m, ok, err := st.BalanceMode()
if err != nil {
log.Printf("ring[%s] read stored balance mode: %v", selfID, err)
return "", false
}
return m, ok
},
func(m string) error {
s, err := st.Settings()
if err != nil {
return err
}
s.BalanceMode = m
return st.UpdateSettings(s)
},
)
// Report the EFFECTIVE mode, not balMode: a stored value may have overridden
// the environment, and a startup line naming the wrong mode is exactly how an
// operator ends up believing releasing is off while it is on.
log.Printf("ring[%s] load balancing %s (source=%s cooldown=%ds interval=%s, self-directed)",
selfID, strings.ToUpper(balancer.Mode()), balancer.ModeSource(), balCooldown, balInterval)
go balancer.Watch(clusterCtx, balInterval)
// Startup reconcile: every node (leader or not) checks its own topology // Startup reconcile: every node (leader or not) checks its own topology
// entries and re-claims any whose per-forward frpc worker process is // entries and re-claims any whose per-forward frpc worker process is
// absent — the topology was recovered via AdoptState but the worker // absent — the topology was recovered via AdoptState but the worker
@ -735,6 +818,22 @@ func retryJoinCluster(ctx context.Context, ring *cluster.Engine, targetAddr, joi
} }
} }
// envFloat reads a numeric environment variable, returning def when unset or
// unparseable. A malformed value must not silently change behaviour, so it falls
// back to the documented default rather than 0.
func envFloat(key string, def float64) float64 {
v := strings.TrimSpace(os.Getenv(key))
if v == "" {
return def
}
f, err := strconv.ParseFloat(v, 64)
if err != nil {
log.Printf("env %s=%q is not a number; using %v", key, v, def)
return def
}
return f
}
// retryRejoinCached is the crash/restart recovery path: parse the persisted // retryRejoinCached is the crash/restart recovery path: parse the persisted
// peer list (JSON of [{addr,key},...]) and try to rejoin via each peer in // peer list (JSON of [{addr,key},...]) and try to rejoin via each peer in
// order. The first peer that accepts wins. Retry the whole list every 10s // order. The first peer that accepts wins. Retry the whole list every 10s

View File

@ -0,0 +1,728 @@
package cluster
import (
"context"
"encoding/json"
"errors"
"strconv"
"strings"
"testing"
"time"
"webui4frpc/internal/store"
)
// mustState round-trips the seed through JSON so the engine under test owns its
// own copy of Nodes/Topology. AdoptState assigns the state by value, but Nodes is
// a slice: handing one seed to several engines would alias one backing array and
// make a mutation on one node visible on the others — a test artefact that
// previously produced a false green here.
func mustState(t *testing.T, s State) *State {
t.Helper()
b, err := json.Marshal(s)
if err != nil {
t.Fatal(err)
}
var out State
if err := json.Unmarshal(b, &out); err != nil {
t.Fatal(err)
}
e := newTestEngine("probe", true)
e.AdoptState(out)
return e.State()
}
// migrateRing builds a 3-node ring where n1 owns every forward, plus n2/n3 idle.
func migrateRing(t *testing.T, ownCount int) (*Engine, *Engine, *Engine) {
t.Helper()
mk := func(id string, leader bool) *Engine {
h := &fakeHandler{load: Load{MemPct: 5, NetPct: 5}}
return NewEngine(id, id+":7500", "u", "p", "0.1.3", nil, h,
func(ctx context.Context, next string, tk *Token) error { return nil },
id+":7500", leader, "")
}
n1, n2, n3 := mk("n1", true), mk("n2", false), mk("n3", false)
seed := State{
LeaderID: "n1",
Nodes: []Node{
{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: Load{Forwards: ownCount, MemPct: 5}},
{ID: "n2", Addr: "n2:7500", Alive: true, Load: Load{MemPct: 5}},
{ID: "n3", Addr: "n3:7500", Alive: true, Load: Load{MemPct: 5}},
},
Topology: map[string]*TopoEntry{},
PendingTasks: map[string]*Task{},
}
for _, e := range []*Engine{n1, n2, n3} {
var own State
b, _ := json.Marshal(seed)
if err := json.Unmarshal(b, &own); err != nil {
t.Fatal(err)
}
e.AdoptState(own)
}
// n1 claims `ownCount` forwards.
for i := 0; i < ownCount; i++ {
n1.state.AddTopology(&Task{
ID: "f" + string(rune('a'+i)),
Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
Link: store.Link{RemotePort: 100 + i},
}, "n1")
}
var snap State
b, _ := json.Marshal(n1.state)
_ = json.Unmarshal(b, &snap)
for _, e := range []*Engine{n1, n2, n3} {
e.AdoptState(snap)
}
return n1, n2, n3
}
// errNotWritable stands in for a failing settings write.
var errNotWritable = errors.New("not writable")
// ---- helpers ----
// balanceRing builds a 3-node ring with `owned` forwards on n1 (the evaluation
// node) and `others` on each of the other two, so a test can set up any
// distribution directly. The replicated Nodes[].Load is made consistent with the
// topology, because that is what a real token round leaves behind.
func balanceRing(t *testing.T, owned, others int) (*Engine, *Engine, *Engine) {
t.Helper()
mk := func(id string, leader bool) *Engine {
h := &fakeHandler{load: Load{MemPct: 5, NetPct: 5}}
return NewEngine(id, id+":7500", "u", "p", "0.1.3", nil, h,
func(ctx context.Context, next string, tk *Token) error { return nil },
id+":7500", leader, "")
}
n1, n2, n3 := mk("n1", true), mk("n2", false), mk("n3", false)
seed := State{
LeaderID: "n1",
Nodes: []Node{
{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: Load{Forwards: owned, MemPct: 5}},
{ID: "n2", Addr: "n2:7500", Alive: true, Load: Load{Forwards: others, MemPct: 5}},
{ID: "n3", Addr: "n3:7500", Alive: true, Load: Load{Forwards: others, MemPct: 5}},
},
Topology: map[string]*TopoEntry{},
PendingTasks: map[string]*Task{},
}
for _, e := range []*Engine{n1, n2, n3} {
var own State
b, _ := json.Marshal(seed)
if err := json.Unmarshal(b, &own); err != nil {
t.Fatal(err)
}
e.AdoptState(own)
}
// Build the whole topology in n1's state, then broadcast it. Adding each
// node's forwards to its OWN state and then re-adopting from n1 would discard
// the others (one state is the source of the broadcast), which is how an
// earlier version of this helper silently produced a 1-1-1 split regardless
// of the arguments.
add := func(owner, prefix string, count, basePort int) {
for i := 0; i < count; i++ {
n1.state.AddTopology(&Task{
ID: prefix + strconv.Itoa(i),
Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
Link: store.Link{RemotePort: basePort + i},
}, owner)
}
}
add("n1", "f", owned, 100)
add("n2", "g", others, 200)
add("n3", "h", others, 300)
var snap State
b, _ := json.Marshal(n1.state)
_ = json.Unmarshal(b, &snap)
for _, e := range []*Engine{n1, n2, n3} {
e.AdoptState(snap)
}
return n1, n2, n3
}
// ---- the rule ----
// TestShareIsTheEvenSplit pins the arithmetic, including the ceiling. An even
// split of 3 over 3 is 1 (not 1.0 floored to 0): with a floor, a node holding
// one of three forwards would still be "over its share" and shed it.
func TestShareIsTheEvenSplit(t *testing.T) {
for _, tc := range []struct{ total, alive, want int }{
{3, 3, 1}, {4, 3, 2}, {6, 2, 3}, {1, 3, 1}, {0, 3, 0}, {7, 3, 3},
} {
got := (tc.total + tc.alive - 1) / tc.alive
if got != tc.want {
t.Errorf("share(%d over %d) = %d, want %d", tc.total, tc.alive, got, tc.want)
}
}
}
// TestOwnerReleasesWhenOverItsShare: 3 forwards on n1, none elsewhere → n1 holds
// 3 against a share of 1 and must give one up.
func TestOwnerReleasesWhenOverItsShare(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
pick, p, blocked := b.assess(1000)
if pick == nil {
t.Fatalf("n1 holds 3 of 3 forwards and should release one (blocked=%q)", blocked)
}
if blocked != "" {
t.Errorf("nothing should block the first release, got %q", blocked)
}
if p.mine != 3 || p.share != 1 || p.total != 3 || p.alive != 3 {
t.Errorf("pool = %+v, want mine=3 share=1 total=3 alive=3", p)
}
if pick.OwnerID != "n1" {
t.Errorf("must release a forward n1 owns, got owner %q", pick.OwnerID)
}
}
// TestReleasingStopsAtTheShare is the bound that keeps the rule from emptying a
// node: after shedding one, n1 holds 1 of 3 over 3 nodes, which is no longer more
// than the share, and it must stop.
func TestReleasingStopsAtTheShare(t *testing.T) {
n1, _, _ := balanceRing(t, 1, 1) // even 1-1-1 split
b := NewBalancer(n1, 0)
pick, p, blocked := b.assess(1000)
if pick != nil {
t.Errorf("an even 1-1-1 split must not release anything, got %s:%d (blocked=%q)",
pick.Local.Name, pick.Link.RemotePort, blocked)
}
if p.mine != 1 || p.share != 1 {
t.Errorf("pool = %+v, want mine=1 share=1", p)
}
}
// TestSingleNodeNeverReleases: a standalone node is the whole cluster, so its
// share is everything it holds. Without this the node would shed forwards onto
// nobody (they would sit unclaimed until it re-claimed them).
func TestSingleNodeNeverReleases(t *testing.T) {
e := newTestEngine("solo", true)
e.state.Nodes = []Node{{ID: "solo", Addr: "solo:7500", Alive: true, IsLeader: true}}
for i := 0; i < 5; i++ {
e.state.AddTopology(&Task{
ID: "f" + strconv.Itoa(i), Local: store.Local{Name: "web"},
Remote: store.Remote{Name: "frps1"}, Link: store.Link{RemotePort: 100 + i},
}, "solo")
}
b := NewBalancer(e, 0)
if pick, _, blocked := b.assess(1000); pick != nil {
t.Errorf("a single-node ring must not release; got blocked=%q", blocked)
}
}
// TestBelowFloorNeverReleases: two forwards on a node is the floor. A node
// holding exactly two at an even split of two must keep both — one release per
// node per pass is the granularity, and shedding below the floor would empty it.
func TestBelowFloorNeverReleases(t *testing.T) {
n1, _, _ := balanceRing(t, 2, 2) // 6 total over 3 nodes → share 2
b := NewBalancer(n1, 0)
if pick, p, blocked := b.assess(1000); pick != nil {
t.Errorf("2 owned against share %d must not release (blocked=%q, mine=%d)",
p.share, blocked, p.mine)
}
}
// TestNeverReleasesAStoppedForward: a stopped forward's entry is the carrier of
// the user's stop decision. Releasing it would hand another node a STOPPED
// forward as if it were work to run.
func TestNeverReleasesAStoppedForward(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
// Stop two of n1's three; the only live one must be chosen.
for _, te := range n1.state.ForwardsOwnedBy("n1") {
if te.Link.RemotePort == 100 || te.Link.RemotePort == 101 {
n1.state.UpdateTopologyDisabled("web", "frps1", te.Link.RemotePort, true)
}
}
b := NewBalancer(n1, 0)
pick, _, _ := b.assess(1000)
if pick == nil {
t.Fatal("n1 still owns one live forward and is over its share")
}
if pick.Link.RemotePort != 102 {
t.Errorf("released port %d; a stopped forward (100/101) must never be chosen",
pick.Link.RemotePort)
}
// And with every forward stopped there is nothing to release.
n1.state.UpdateTopologyDisabled("web", "frps1", 102, true)
if pick, _, blocked := b.assess(1000); pick != nil {
t.Errorf("all forwards stopped: nothing may be released (blocked=%q)", blocked)
}
}
// TestCandidateChoiceIsDeterministic: the same state must pick the same forward,
// so the decision is explainable and a re-evaluation does not churn a different
// connection each tick.
func TestCandidateChoiceIsDeterministic(t *testing.T) {
n1, _, _ := balanceRing(t, 4, 0)
a := releaseCandidate(n1.State(), "n1")
c := releaseCandidate(n1.State(), "n1")
if a == nil || c == nil {
t.Fatal("expected a candidate")
}
if a.TaskID != c.TaskID || a.Link.RemotePort != c.Link.RemotePort {
t.Errorf("candidate changed between identical evaluations: %s:%d vs %s:%d",
a.TaskID, a.Link.RemotePort, c.TaskID, c.Link.RemotePort)
}
if a.Link.RemotePort != 100 {
t.Errorf("expected the lowest port (100), got %d", a.Link.RemotePort)
}
}
// ---- the runtime signal that makes this safe ----
// TestRevokeLeavesACarrierAndReleaseRemovesIt is the reason the release path has
// to be a distinct task kind.
//
// A user stop keeps the topology entry as the carrier of the disabled flag, so
// the entry is STILL THERE and still attributed to the owner. A rebalance is the
// opposite: the entry must go, because (a) the node's load is derived from
// owned entries so it cannot fall while the entry exists, and (b) every node's
// duplicate guard refuses a creation task for a forward that already has an owner
// — so the re-published task would be dropped by everyone and the forward
// stranded on the node that gave it up.
func TestRevokeLeavesACarrierAndReleaseRemovesIt(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
// A stop marks the entry and keeps it.
n1.state.UpdateTopologyDisabled("web", "frps1", 100, true)
tk := &Task{Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
Link: store.Link{RemotePort: 100}}
if owner := n1.state.TopologyOwner(tk); owner != "n1" {
t.Fatalf("a stop must keep the entry as the flag's carrier, owner=%q", owner)
}
if got := len(n1.state.ForwardsOwnedBy("n1")); got != 3 {
t.Fatalf("a stop must not change the owner's forward count, got %d", got)
}
// A release removes the entry (leaving the other two alone).
if !n1.RemoveTopologyEntry("web", "frps1", 100) {
t.Fatal("release must remove the entry")
}
if owner := n1.state.TopologyOwner(tk); owner != "" {
t.Errorf("after a release the forward must have no owner, got %q", owner)
}
if got := len(n1.state.ForwardsOwnedBy("n1")); got != 2 {
t.Errorf("a release must reduce the owner's count, got %d", got)
}
if got := len(n1.state.TopologyList()); got != 2 {
t.Errorf("the other forwards must be untouched, got %d entries", got)
}
}
// TestReleaseTaskExecutesOnlyOnTheOwner: like a revoke, a release is
// owner-directed. Executing it elsewhere would be an attempt to move somebody
// else's load, which cannot work — the owner's own count is what must fall.
func TestReleaseTaskExecutesOnlyOnTheOwner(t *testing.T) {
n1, n2, _ := balanceRing(t, 3, 0)
released := map[string]int{}
for _, e := range []*Engine{n1, n2} {
h := e.Handler.(*fakeHandler)
h.release = func(ctx context.Context, tk *Task) error {
released[e.ID]++
return nil
}
}
tk := n1.ReleaseTask(store.Local{Name: "web"}, store.Remote{Name: "frps1"},
store.Link{RemotePort: 100})
if tk == nil {
t.Fatal("ReleaseTask returned nil")
}
// Each Engine keeps its OWN state, so a task published on n1 is not visible
// to n2 until a token round carries it. Broadcast it the way the ring does.
var carry State
cb, _ := json.Marshal(n1.state)
_ = json.Unmarshal(cb, &carry)
n2.AdoptState(carry)
n2.state.PendingTasks[tk.ID] = tk
// n2 is the lightest node and does NOT own the forward: it must leave the
// task alone.
if err := n2.runCommands(context.Background(), &Token{}); err != nil {
t.Fatalf("n2 runCommands: %v", err)
}
if released["n2"] != 0 {
t.Errorf("n2 released %d forward(s); a release is owner-directed only", released["n2"])
}
if _, still := n2.state.PendingTasks[tk.ID]; !still {
t.Error("n2 consumed a release it does not own; it must ride the token to the owner")
}
// n1 owns it: it must release, drop the entry, and re-publish the forward.
if err := n1.runCommands(context.Background(), &Token{}); err != nil {
t.Fatalf("n1 runCommands: %v", err)
}
if released["n1"] != 1 {
t.Fatalf("the owner released %d times, want 1", released["n1"])
}
if owner := n1.state.TopologyOwner(&Task{
Local: store.Local{Name: "web"}, Remote: store.Remote{Name: "frps1"},
Link: store.Link{RemotePort: 100},
}); owner != "" {
t.Errorf("the released forward still has owner %q", owner)
}
// And it is back in the ring as an ordinary creation task.
var republished *Task
for _, p := range n1.state.PendingList() {
if !p.Release && !p.Revoke && !p.Restart &&
p.Link.RemotePort == 100 && p.Local.Name == "web" {
republished = p
}
}
if republished == nil {
t.Fatal("the released forward was not re-published as an unclaimed task")
}
if republished.Link.Disabled {
t.Error("the re-published forward must be ENABLED, not marked stopped")
}
}
// ---- dry-run / off ----
// TestDryRunReleasesNothing: dry-run is the default, and its whole value is that
// it can run on a live cluster without disturbing anything.
func TestDryRunReleasesNothing(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceDryRun)
msg := b.Step(1000)
if !strings.Contains(msg, "would release") {
t.Fatalf("dry-run should report what it would do, got %q", msg)
}
if b.InFlight() {
t.Error("dry-run started a release")
}
for _, tk := range n1.state.PendingList() {
if tk.Release {
t.Error("dry-run published a release task")
}
}
if got := len(n1.state.TopologyList()); got != 3 {
t.Errorf("dry-run changed the topology: %d entries, want 3", got)
}
// A second identical tick must not repeat the same line (log hygiene).
if again := b.Step(1010); again != "" {
t.Errorf("dry-run repeated an unchanged decision: %q", again)
}
}
// TestModeOffDoesNothing reinforces that OFF means off, not "dry-run".
func TestModeOffDoesNothing(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceOff)
if msg := b.Step(1000); msg != "" {
t.Errorf("mode off reported %q", msg)
}
if b.InFlight() {
t.Error("mode off started a release")
}
// A typo must not enable releasing.
b.SetMode("yes-please")
if b.Mode() != BalanceDryRun {
t.Errorf("an unknown mode must fall back to dry-run, got %q", b.Mode())
}
}
// ---- end-to-end through the ring machinery ----
// TestOwnerReleaseThenRingClaimsElsewhere is the whole design in one test: the
// owner gives the forward up, and the ordinary lowest-load claim then places it
// on an idle node — exactly what the self-directed rule depends on.
func TestOwnerReleaseThenRingClaimsElsewhere(t *testing.T) {
n1, n2, n3 := balanceRing(t, 3, 0)
// n2 and n3 must look idle to the claim path (they do: Forwards=0).
claimed := map[string]int{}
for _, e := range []*Engine{n1, n2, n3} {
h := e.Handler.(*fakeHandler)
h.claim = func(ctx context.Context, tk *Task) error {
claimed[e.ID]++
return nil
}
}
b := NewBalancer(n1, 0)
b.SetMode(BalanceOn)
// n1 is over its share → it releases.
msg := b.Step(1000)
if !strings.Contains(msg, "releasing") {
t.Fatalf("expected a release, got %q", msg)
}
if err := n1.runCommands(context.Background(), &Token{}); err != nil {
t.Fatalf("n1 runCommands: %v", err)
}
// Find the task the owner re-published for the forward it gave up.
var tk *Task
for _, p := range n1.state.PendingList() {
if !p.Release && p.Link.RemotePort == 100 && p.Local.Name == "web" {
tk = p
}
}
if tk == nil {
t.Fatal("the owner did not re-publish the forward it released")
}
if got := len(n1.state.ForwardsOwnedBy("n1")); got != 2 {
t.Fatalf("n1 should now own 2, got %d", got)
}
// The forward is unclaimed; the lowest node claims it. n2 is the lightest
// (Forwards=0 and the same mem/net), so it should take it. The task reaches
// n2 on the token — each engine holds its own state, so carry it over the way
// a real round does.
var carried State
cb, _ := json.Marshal(n1.state)
_ = json.Unmarshal(cb, &carried)
n2.AdoptState(carried)
n2.state.PendingTasks[tk.ID] = tk
if err := n2.runCommands(context.Background(), &Token{}); err != nil {
t.Fatalf("n2 runCommands: %v", err)
}
if claimed["n2"] == 0 {
t.Fatal("the idle node did not claim the released forward")
}
// And n1 no longer owns it.
for _, te := range n1.state.ForwardsOwnedBy("n1") {
if te.Link.RemotePort == 100 {
t.Error("n1 still owns the forward it released")
}
}
// The balancer sees the release land and stops holding the release state.
if msg := b.Step(1100); !strings.Contains(msg, "gave up") {
t.Errorf("expected the release to be confirmed, got %q", msg)
}
if b.InFlight() {
t.Error("the release should be complete")
}
}
// TestCooldownBoundsTheChurn: a release costs a live connection, so one node may
// not shed again until the cooldown elapses — even if it is still over its share.
func TestCooldownBoundsTheChurn(t *testing.T) {
n1, _, _ := balanceRing(t, 4, 0)
b := NewBalancer(n1, 600) // 10 minutes
b.SetMode(BalanceOn)
if msg := b.Step(1000); !strings.Contains(msg, "releasing") {
t.Fatalf("first release expected, got %q", msg)
}
n1.RemoveTopologyEntry("web", "frps1", 100) // land it
if msg := b.Step(1010); !strings.Contains(msg, "gave up") {
t.Fatalf("expected confirmation, got %q", msg)
}
// Still over its share (3 of 4 over 3 → share 2), but inside the cooldown.
if pick, _, blocked := b.assess(1020); pick == nil || blocked != "cooldown" {
t.Errorf("expected the second release to be blocked by cooldown, got blocked=%q", blocked)
}
if b.InFlight() {
t.Error("cooldown must not leave a release in flight")
}
// Past the cooldown it may act again.
if pick, _, blocked := b.assess(1000 + 601); pick == nil || blocked != "" {
t.Errorf("past the cooldown the release should be allowed, got blocked=%q", blocked)
}
}
// TestReleaseDeadlineClearsAStuckConfirmation: if the release never lands (leader
// change, ring join), the node must not wedge forever.
func TestReleaseDeadlineClearsAStuckConfirmation(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceOn)
_ = b.Step(1000) // release published, entry NOT removed (nothing executed it)
if !b.InFlight() {
t.Fatal("precondition: a release should be in flight")
}
if msg := b.Step(1000 + releaseDeadlineSec + 1); msg != "" {
t.Errorf("a timed-out release should clear silently, got %q", msg)
}
if b.InFlight() {
t.Error("the release state should have been cleared after the deadline")
}
}
// ---- read-only report ----
// TestReportDoesNotMutate — polling the endpoint must not start, delay or cancel
// a release. A read-only surface that perturbs what it reports is worse than none.
func TestReportDoesNotMutate(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 600)
b.SetMode(BalanceOn)
for i := 0; i < 5; i++ {
rep := b.Report()
if rep.SelfForwards != 3 {
t.Fatalf("report changed the node's forward count: %d", rep.SelfForwards)
}
if rep.Mode != BalanceOn {
t.Fatalf("report changed the mode: %q", rep.Mode)
}
}
if b.InFlight() || b.lastReleaseAt != 0 {
t.Error("polling the report started a release")
}
if rep := b.Report(); rep.WouldRelease == nil {
t.Error("the report should still surface the pending decision")
}
}
// TestReportExplainsWhyNothingHappens pins the field that answers the question
// the log cannot: an over-share node that is holding back must say why.
func TestReportExplainsWhyNothingHappens(t *testing.T) {
n1, _, _ := balanceRing(t, 4, 0)
b := NewBalancer(n1, 600)
b.SetMode(BalanceOn)
rep := b.Report()
if !rep.SelfOverShare {
t.Fatalf("4 of 4 forwards over 3 nodes should be over the share (share=%d)", rep.Share)
}
if rep.WouldRelease == nil {
t.Fatal("an over-share node must surface the forward it would give up")
}
if rep.BlockedBy != "" {
t.Errorf("nothing should block the first release, got %q", rep.BlockedBy)
}
// Inside the cooldown the same node reports the block. Report() reads the
// real clock, so set lastReleaseAt relative to now rather than to a fake one.
b.lastReleaseAt = time.Now().Unix() - 10
n1.RemoveTopologyEntry("web", "frps1", 100)
rep = b.Report()
if rep.BlockedBy != "cooldown" {
t.Errorf("after a release the node should report a cooldown block, got %q", rep.BlockedBy)
}
}
// TestReportOnBalancedNodeIsQuiet: the healthy case must be a real answer, not an
// empty one — otherwise "balanced" and "balancing is switched off" look identical.
func TestReportOnBalancedNodeIsQuiet(t *testing.T) {
n1, _, _ := balanceRing(t, 1, 1)
b := NewBalancer(n1, 0)
b.SetMode(BalanceOn)
rep := b.Report()
if rep.WouldRelease != nil {
t.Errorf("a balanced node reported a release: %+v", rep.WouldRelease)
}
if rep.SelfOverShare {
t.Error("1 of 3 over 3 nodes is not over the share")
}
if rep.TotalForwards != 3 || rep.AliveNodes != 3 || rep.Share != 1 {
t.Errorf("report pool = total %d alive %d share %d, want 3/3/1",
rep.TotalForwards, rep.AliveNodes, rep.Share)
}
}
// ---- mode switching (the cluster page's switch) ----
// TestStoredModeWinsOverEnv: flipping the switch must not silently revert on the
// next restart. The environment is only the bootstrap default; once an operator
// has chosen a mode it is persisted and wins.
func TestStoredModeWinsOverEnv(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode("dry-run") // the env default
var saved string
b.SetModePersistence(
func() (string, bool) { return "on", true },
func(m string) error { saved = m; return nil },
)
if b.Mode() != BalanceOn {
t.Fatalf("a stored mode must win over the environment default, got %q", b.Mode())
}
if b.ModeSource() != "stored" {
t.Errorf("ModeSource = %q, want stored", b.ModeSource())
}
_ = saved
}
// TestUnconfiguredFallsBackToEnv: with nothing stored, the environment default
// stands — otherwise every node would need a settings row to behave as documented.
func TestUnconfiguredFallsBackToEnv(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode("off")
b.SetModePersistence(func() (string, bool) { return "", false }, func(string) error { return nil })
if b.Mode() != BalanceOff {
t.Errorf("with nothing stored the env default must stand, got %q", b.Mode())
}
if b.ModeSource() != "stored" {
// persist is installed, so the node CAN be controlled
t.Logf("ModeSource=%q", b.ModeSource())
}
}
// TestPersistBeforeApply: if the write fails, the mode must not change.
// Applying first would leave the node running a mode it cannot reproduce after a
// restart — the switch would lie.
func TestPersistBeforeApply(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceDryRun)
b.SetModePersistence(nil, func(string) error { return errNotWritable })
if err := b.SetModeFromUI(BalanceOn); err == nil {
t.Fatal("a failed save must surface an error")
}
if b.Mode() != BalanceDryRun {
t.Errorf("a failed save must leave the mode unchanged, got %q", b.Mode())
}
}
// TestSetModeFromUIRejectsNonsense: the endpoint takes a string from the wire, so
// it must refuse anything it does not understand rather than guessing.
func TestSetModeFromUIRejectsNonsense(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceDryRun)
if err := b.SetModeFromUI("yes"); err == nil {
t.Fatal("an unknown mode must be rejected, not coerced")
}
if b.Mode() != BalanceDryRun {
t.Errorf("a rejected mode must not change anything, got %q", b.Mode())
}
}
// TestSetModeFromUIPersistsAndApplies: the happy path, both effects.
func TestSetModeFromUIPersistsAndApplies(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
b.SetMode(BalanceDryRun)
var saved string
b.SetModePersistence(nil, func(m string) error { saved = m; return nil })
if err := b.SetModeFromUI(BalanceOn); err != nil {
t.Fatalf("SetModeFromUI: %v", err)
}
if saved != BalanceOn {
t.Errorf("saved %q, want on", saved)
}
if b.Mode() != BalanceOn {
t.Errorf("mode = %q, want on", b.Mode())
}
if !b.Report().CanControl {
t.Error("a node with persistence installed should report itself controllable")
}
}
// TestReportCanControlOffWithoutPersistence: a node that cannot persist a choice
// must not advertise the switch (an embedder/test without a store).
func TestReportCanControlOffWithoutPersistence(t *testing.T) {
n1, _, _ := balanceRing(t, 3, 0)
b := NewBalancer(n1, 0)
if b.Report().CanControl {
t.Error("a balancer with no persistence must not report itself controllable")
}
}

View File

@ -68,6 +68,19 @@ type Task struct {
// duplicate-claim guard would discard it. The owner applies this one // duplicate-claim guard would discard it. The owner applies this one
// unconditionally: re-mark the entry enabled and (re)spawn the worker. // unconditionally: re-mark the entry enabled and (re)spawn the worker.
Restart bool `json:"restart,omitempty"` Restart bool `json:"restart,omitempty"`
// Release marks a REBALANCING task: the OWNER of the forward gives it up so
// the ring can place it elsewhere. Only the owning node may act on it — it is
// that node's own load which has to fall, and only it can make that true (its
// topology entry is local bookkeeping no peer may write).
//
// It is deliberately NOT the same thing as a user stop (Revoke). A stop keeps
// the topology entry as the carrier of the disabled flag and flags the link
// stopped; a release REMOVES the entry and leaves the link enabled, because
// the forward is about to run somewhere else. Marking it stopped instead would
// make the re-published creation task look like a duplicate of an existing
// forward, and every node would refuse to claim it — stranding the forward on
// the node that just gave it up.
Release bool `json:"release,omitempty"`
// RemoveNode: node ID to remove from the ring; the node self-removes when // RemoveNode: node ID to remove from the ring; the node self-removes when
// the command reaches it via the token. // the command reaches it via the token.
RemoveNode string `json:"removeNode,omitempty"` RemoveNode string `json:"removeNode,omitempty"`

View File

@ -0,0 +1,505 @@
package cluster
import (
"context"
"fmt"
"log"
"sort"
"strconv"
"strings"
"time"
"webui4frpc/internal/store"
)
// The rebalance rule, deliberately simple and self-limiting:
//
// release one forward when Forwards >= minForwardsToShare AND Forwards > share
//
// where share = ceil(totalForwards / aliveNodes) — the even split.
//
// Every node runs this check about its OWN forwards. That is the whole point:
// a node's load only actually falls when its own topology entry goes away, so
// only the owner can produce the state change that makes the ring hand the
// forward to somebody else. A coordinator on another machine can see the
// imbalance but cannot make it true; it would have to reach in and mutate the
// owner's bookkeeping, which is exactly the sort of cross-node write this ring
// avoids everywhere else (revoke and restart are owner-directed for the same
// reason).
//
// The rule is self-limiting in both directions, so it needs no upper bound on
// how many forwards a node may shed:
//
// - It stops at the average. 3 forwards over 3 nodes: the owner holds 2 > 1,
// releases one, now holds 1 which is not > 1, and stops. Without the "> share"
// bound an owner would shed EVERY forward it has.
// - It cannot empty a node. A node holding fewer than minForwardsToShare never
// releases, so there is always something left to serve.
//
// A release costs a live connection: stopping the worker drops whatever was
// flowing through the forward, and the new owner re-dials the local service.
// Because the rule above already requires a real imbalance, pacing is handled by
// a per-node cooldown (cooldownSec) rather than a "how long has this lasted"
// window — there is no transient to smooth over, only a churn rate to bound.
const (
// minForwardsToShare: never release below this many owned forwards.
minForwardsToShare = 2
// balanceCooldownSec: minimum gap between two releases by one node. One
// release per 10 minutes per node is a trickle; it is what stops a cluster
// whose loads oscillate around the split point from restarting workers in a
// loop.
balanceCooldownSec = 600
// releaseDeadlineSec: how long to wait for a published release to be
// executed by its owner (us) before giving up on confirming it. A token
// round is ~2s, so this is generous; it exists so a release lost to a
// leader change or a joined ring cannot wedge the node forever.
releaseDeadlineSec = 120
)
// Balance modes.
const (
BalanceOff = "off"
BalanceDryRun = "dry-run"
BalanceOn = "on"
)
// pool is the replicated view the rule works from.
type pool struct {
alive int // nodes that can take a forward
mine int // forwards THIS node owns (live count, not the replicated one)
total int // forwards across the ring
share int // ceil(total/alive) — the even split
}
// releaseState is one release this node has published and is waiting to see
// leave its own topology.
type releaseState struct {
local, remote string
port int
startedAt int64
}
// Balancer lets each node shed its own share of the cluster's forwards.
//
// It publishes only ONE kind of thing: a release of a forward it owns. There is
// no targeted hand-off and no coordinator, because "drop it and let the ring's
// normal lowest-load claim pick it up" is already the ring's only mechanism for
// moving a forward — and it is the mechanism that, unlike a coordinated move,
// cannot be undone by the node that gave the forward up (the claim path runs
// after the node's load has already fallen).
type Balancer struct {
engine *Engine
mode string
// cooldownSec is the minimum gap between two releases by this node.
cooldownSec int64
// releasing is the forward this node gave up and is waiting to see leave.
// While it is set, no second release is started: one at a time keeps the
// connection churn bounded and the log readable.
releasing *releaseState
// lastReleaseAt is when the previous release was published (0 = never).
lastReleaseAt int64
// reported remembers the last dry-run candidate so the observe-only mode
// logs a change of mind rather than the same line every tick.
reported string
// persist saves a mode change so it survives a restart (nil = env-only).
persist func(string) error
// savedMode is the last mode this node stored, so the report can say whether
// the active mode came from the environment or from an operator's choice.
savedMode string
}
// NewBalancer builds a balancer. It starts in observe-only mode: releasing a
// forward drops a live connection, so acting on the ring is opt-in.
func NewBalancer(e *Engine, cooldownSec int64) *Balancer {
b := &Balancer{engine: e, mode: BalanceDryRun, cooldownSec: cooldownSec}
if b.cooldownSec <= 0 {
b.cooldownSec = balanceCooldownSec
}
return b
}
// SetMode selects off / dry-run / on. An unrecognised value is treated as
// dry-run — a typo must not be able to turn releasing on.
//
// Only a change is logged, and the reason is recorded, because SetMode is now
// called both at boot (from the environment) and at runtime (from the cluster
// page): a one-line "why did this just start moving my forwards" answer is the
// difference between a switch an operator dares to use and one they don't.
func (b *Balancer) SetMode(m string) {
if m != BalanceOff && m != BalanceDryRun && m != BalanceOn {
if m != "" {
log.Printf("ring[%s] unknown balance mode %q; using %s", b.engine.ID, m, BalanceDryRun)
}
m = BalanceDryRun
}
if b.mode == m {
return
}
b.mode = m
b.savedMode = m
log.Printf("ring[%s] load balancing set to %s", b.engine.ID, strings.ToUpper(m))
}
// Mode returns the active mode.
func (b *Balancer) Mode() string { return b.mode }
// ModeSource reports where the active mode came from, for the report.
//
// "stored" means an operator chose it on the cluster page and it will survive a
// restart; "env" means nothing was ever chosen here and the environment default
// (or the built-in dry-run) applies. It is NOT "can this node save a choice" —
// that is CanControl, and conflating the two made every node with a store claim
// an operator had set it.
func (b *Balancer) ModeSource() string {
if b.savedMode != "" {
return "stored"
}
return "env"
}
// SetModePersistence installs the callback that saves a mode change so it
// survives a restart, and applies any previously saved value.
//
// The environment is the BOOTSTRAP default; once an operator flips the switch on
// the cluster page that stored value wins, otherwise the page would appear to
// work and then silently revert on the next restart.
func (b *Balancer) SetModePersistence(load func() (string, bool), save func(string) error) {
b.persist = save
if load == nil {
return
}
if m, ok := load(); ok && m != "" {
switch m {
case BalanceOff, BalanceDryRun, BalanceOn:
if b.mode != m {
log.Printf("ring[%s] load balancing mode %s loaded from settings", b.engine.ID, m)
}
b.mode = m
b.savedMode = m
default:
log.Printf("ring[%s] ignoring unknown stored balance mode %q", b.engine.ID, m)
}
}
}
// SetModeFromUI changes the mode at runtime and persists it.
//
// The order matters: persist FIRST, then apply. If the write fails nothing
// changes, so the switch cannot report success while behaving differently after
// the next restart.
func (b *Balancer) SetModeFromUI(m string) error {
if m != BalanceOff && m != BalanceDryRun && m != BalanceOn {
return fmt.Errorf("未知模式 %q", m)
}
if b.persist != nil {
if err := b.persist(m); err != nil {
return fmt.Errorf("保存负载均衡模式失败: %w", err)
}
}
b.SetMode(m)
return nil
}
// SetCooldown overrides the per-node release cooldown.
func (b *Balancer) SetCooldown(sec int64) {
if sec > 0 {
b.cooldownSec = sec
}
}
// InFlight reports whether a release is awaiting confirmation.
func (b *Balancer) InFlight() bool { return b.releasing != nil }
// pool computes the ring-wide view this node derives for itself.
//
// Our own count is taken LIVE from the topology rather than from the replicated
// Nodes[].Load, because Nodes[].Load is only refreshed when a token arrives and
// the decision must see the effect of the release immediately — otherwise a node
// would keep believing it is over its share and shed another forward each round.
func (b *Balancer) pool() pool {
st := &b.engine.state
p := pool{mine: len(st.ForwardsOwnedBy(b.engine.ID))}
for i := range st.Nodes {
n := &st.Nodes[i]
if !n.Alive {
continue
}
p.alive++
if n.ID == b.engine.ID {
p.total += p.mine
continue
}
p.total += n.Load.Forwards
}
if p.alive > 0 {
p.share = (p.total + p.alive - 1) / p.alive
}
return p
}
// assess answers "should this node give one forward up, and which one?" together
// with the facts behind the answer, so the read-only report can explain a node
// that is over its share but not acting. It never mutates anything.
//
// blocked is a short reason: "single-node", "below-min", "balanced",
// "nothing-active", "cooldown" or "in-flight".
func (b *Balancer) assess(now int64) (pick *TopoEntry, p pool, blocked string) {
p = b.pool()
if !b.engine.IsMember() {
return nil, p, "single-node"
}
if p.mine < minForwardsToShare {
return nil, p, "below-min"
}
if p.mine <= p.share {
return nil, p, "balanced"
}
pick = releaseCandidate(&b.engine.state, b.engine.ID)
if pick == nil {
return nil, p, "nothing-active"
}
// The guards below do not change the verdict ("this node is over its
// share"), only whether now is the moment; the report shows both.
if b.releasing != nil {
return pick, p, "in-flight"
}
if b.lastReleaseAt > 0 && now-b.lastReleaseAt < b.cooldownSec {
return pick, p, "cooldown"
}
return pick, p, ""
}
// releaseCandidate picks which of this node's forwards to give up.
//
// Deterministic (lowest remote port, then task id) so the choice is explainable
// and repeatable: a node that re-evaluates the same state picks the same
// forward. Stopped forwards are skipped — their entry is the carrier of a user's
// stop decision, and releasing one would hand a STOPPED forward to another node
// as if it were work to run.
func releaseCandidate(s *State, owner string) *TopoEntry {
var mine []*TopoEntry
for _, te := range s.TopologyList() {
if te.OwnerID != owner || !te.Active || te.Link.Disabled {
continue
}
mine = append(mine, te)
}
if len(mine) == 0 {
return nil
}
sort.Slice(mine, func(i, j int) bool {
if mine[i].Link.RemotePort != mine[j].Link.RemotePort {
return mine[i].Link.RemotePort < mine[j].Link.RemotePort
}
return mine[i].TaskID < mine[j].TaskID
})
return mine[0]
}
// landed reports whether the forward this node released has actually left it.
//
// The test is "no longer owned by me", NOT "no entry exists": the ring's normal
// claim picks the forward up within a round or two, so an existence test would
// see the new owner's entry and wait forever for a state that has already
// happened.
func (b *Balancer) landed() bool {
m := b.releasing
if m == nil {
return true
}
owner := b.engine.state.TopologyOwner(&Task{
Local: store.Local{Name: m.local},
Remote: store.Remote{Name: m.remote},
Link: store.Link{RemotePort: m.port},
})
return owner != b.engine.ID
}
// Step advances the balancer by one cycle and returns a human-readable
// description when it did (or, in dry-run, would have done) something.
func (b *Balancer) Step(now int64) string {
if b.releasing != nil {
if b.landed() {
done := b.releasing
b.releasing = nil
return "gave up " + done.local + ":" + strconv.Itoa(done.port) + " for rebalancing"
}
if now-b.releasing.startedAt > releaseDeadlineSec {
log.Printf("ring[%s] release of %s:%d not confirmed within %ds; clearing (forward is still ours)",
b.engine.ID, b.releasing.local, b.releasing.port, releaseDeadlineSec)
b.releasing = nil
}
return ""
}
pick, p, blocked := b.assess(now)
if pick == nil {
b.reported = ""
return ""
}
desc := pick.Local.Name + ":" + strconv.Itoa(pick.Link.RemotePort) +
" (" + strconv.Itoa(p.mine) + " owned > " + strconv.Itoa(p.share) + " share of " +
strconv.Itoa(p.total) + " over " + strconv.Itoa(p.alive) + " nodes)"
switch b.mode {
case BalanceOff:
return ""
case BalanceDryRun:
// Log a change of mind, not the same line every tick.
if b.reported == desc {
return ""
}
b.reported = desc
if blocked != "" {
return "would release " + desc + " [blocked: " + blocked + "]"
}
return "would release " + desc
}
// Mode is ON. Blocked reasons still stop us.
if blocked != "" {
return ""
}
t := b.engine.ReleaseTask(pick.Local, pick.Remote, pick.Link)
if t == nil {
return ""
}
b.releasing = &releaseState{
local: pick.Local.Name, remote: pick.Remote.Name,
port: pick.Link.RemotePort, startedAt: now,
}
b.lastReleaseAt = now
b.reported = ""
return "releasing " + desc
}
// Watch drives Step on a ticker until ctx is cancelled.
//
// It runs on EVERY node, not just the leader. Balancing here is not a cluster
// decision being delegated to one node — it is each node applying the same
// replicated rule to its own forwards. Two nodes never contend for the same
// forward because a node can only release what it owns.
//
// The first check happens after one interval rather than immediately: a node
// that has just booted adopts the ring's topology and its own load in the first
// token round, and deciding before that would act on a half-known pool.
func (b *Balancer) Watch(ctx context.Context, interval time.Duration) {
tick := time.NewTicker(interval)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
}
if msg := b.Step(time.Now().Unix()); msg != "" {
log.Printf("ring[%s] balance: %s", b.engine.ID, msg)
}
}
}
// ---- read-only report for the API/UI ----
// BalanceReport is what this node thinks about balancing right now.
//
// It is per-node by nature: every member has its own answer, and the answer that
// matters is the one on the node doing (or refusing to do) the releasing. The
// peer list is included because "am I over my share?" is only meaningful next to
// what everyone else holds.
type BalanceReport struct {
Mode string `json:"mode"` // off | dry-run | on
ModeSource string `json:"modeSource"` // env | stored
Enabled bool `json:"enabled"` // mode != off
DryRun bool `json:"dryRun"` // mode != on
// CanControl: this node offers the mode switch. False for a read-only
// principal, so the server does not advertise a control the UI gates anyway.
CanControl bool `json:"canControl"`
IsMember bool `json:"isMember"`
Self string `json:"self"`
SelfForwards int `json:"selfForwards"`
Share int `json:"share"`
TotalForwards int `json:"totalForwards"`
AliveNodes int `json:"aliveNodes"`
SelfOverShare bool `json:"selfOverShare"`
// WouldRelease is what this node would give up (present whenever it is over
// its share, including while blocked — see BlockedBy).
WouldRelease *BalanceReleasePlan `json:"wouldRelease,omitempty"`
// Releasing is the release this node has published and is awaiting.
Releasing *BalanceReleasePlan `json:"releasing,omitempty"`
// BlockedBy explains why an over-share node is not acting yet.
BlockedBy string `json:"blockedBy,omitempty"`
CooldownSec int64 `json:"cooldownSec"`
LastReleaseAt int64 `json:"lastReleaseAt,omitempty"`
// Peers: every member's replicated load, for context.
Nodes []BalanceNodeReport `json:"nodes"`
}
type BalanceNodeReport struct {
ID string `json:"id"`
Alive bool `json:"alive"`
IsLeader bool `json:"isLeader"`
Forwards int `json:"forwards"`
MemPct float64 `json:"memPct"`
NetPct float64 `json:"netPct"`
Score float64 `json:"score"`
}
type BalanceReleasePlan struct {
Local string `json:"local"`
Remote string `json:"remote"`
RemotePort int `json:"remotePort"`
Owner string `json:"owner"`
}
// Report renders the current view. It never mutates anything — in particular it
// does NOT touch lastReleaseAt or the release state, so an operator polling this
// cannot start, delay or cancel a release. (A read-only endpoint that perturbs
// the behaviour it reports is worse than no endpoint.)
func (b *Balancer) Report() BalanceReport {
now := time.Now().Unix()
rep := BalanceReport{
Mode: b.mode,
ModeSource: b.ModeSource(),
Enabled: b.mode != BalanceOff,
DryRun: b.mode != BalanceOn,
CanControl: b.persist != nil,
IsMember: b.engine.IsMember(),
Self: b.engine.ID,
CooldownSec: b.cooldownSec,
}
p := b.pool()
rep.SelfForwards, rep.Share = p.mine, p.share
rep.TotalForwards, rep.AliveNodes = p.total, p.alive
rep.SelfOverShare = p.mine > p.share
if pick, _, blocked := b.assess(now); pick != nil {
rep.WouldRelease = &BalanceReleasePlan{
Local: pick.Local.Name, Remote: pick.Remote.Name,
RemotePort: pick.Link.RemotePort, Owner: pick.OwnerID,
}
rep.BlockedBy = blocked
}
if m := b.releasing; m != nil {
rep.Releasing = &BalanceReleasePlan{
Local: m.local, Remote: m.remote, RemotePort: m.port, Owner: b.engine.ID,
}
rep.BlockedBy = "in-flight"
}
if b.lastReleaseAt > 0 {
rep.LastReleaseAt = b.lastReleaseAt
}
for i := range b.engine.state.Nodes {
n := &b.engine.state.Nodes[i]
rep.Nodes = append(rep.Nodes, BalanceNodeReport{
ID: n.ID, Alive: n.Alive, IsLeader: n.IsLeader,
Forwards: n.Load.Forwards, MemPct: n.Load.MemPct, NetPct: n.Load.NetPct,
Score: n.Load.Score(),
})
}
return rep
}

View File

@ -24,6 +24,12 @@ type Handler interface {
// creation path (idempotency guard) and the duplicate-claim guard refuse to // creation path (idempotency guard) and the duplicate-claim guard refuse to
// re-create an existing forward. // re-create an existing forward.
Restart(ctx context.Context, tk *Task) error Restart(ctx context.Context, tk *Task) error
// Release stops this node's worker for a forward it is giving up for
// rebalancing, WITHOUT marking the link stopped. The forward is about to run
// on another node, so a stopped flag would be wrong: every node's duplicate
// guard reads it and would then refuse to claim the re-published task,
// stranding the forward on the node that just gave it up.
Release(ctx context.Context, tk *Task) error
RuntimeLoad() Load RuntimeLoad() Load
} }
@ -406,6 +412,25 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
} }
continue // owned elsewhere — ride to the owner continue // owned elsewhere — ride to the owner
} }
if t.Release {
// Owner-directed, like a revoke: only the node holding the forward
// may give it up. A release executed anywhere else would be an
// attempt to move somebody else's load, which cannot work — the
// owner's own count is what has to fall, and its topology entry is
// local bookkeeping no peer may write.
//
// A release whose forward is already gone is an idempotent no-op
// consumed by the lowest node, so it cannot ride forever if its
// owner departed first.
if owner := e.state.TopologyOwner(t); owner == e.ID {
target = t
break
} else if owner == "" && selfIsLowest {
target = t
break
}
continue // owned elsewhere — ride to the owner
}
if t.Restart { if t.Restart {
// Owner-directed, exactly like a revoke: only the node holding the // Owner-directed, exactly like a revoke: only the node holding the
// forward may act. Selecting it anywhere else would let a non-owner // forward may act. Selecting it anywhere else would let a non-owner
@ -538,7 +563,11 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
// exists precisely to re-activate a forward whose entry is still present // exists precisely to re-activate a forward whose entry is still present
// (a stop keeps the entry as the flag's carrier), so applying the guard // (a stop keeps the entry as the flag's carrier), so applying the guard
// would swallow every re-enable and leave the forward stopped forever. // would swallow every re-enable and leave the forward stopped forever.
if !claimed.Restart { // A RELEASE is the second deliberate exception, for the same reason as a
// restart: it is addressed to the node that ALREADY owns the forward, so
// the entry is still there by definition. Applying the guard would drop
// every release and a node could never shed anything.
if !claimed.Restart && !claimed.Release {
if owner := e.state.TopologyOwner(claimed); owner != "" { if owner := e.state.TopologyOwner(claimed); owner != "" {
log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s", log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s",
e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name, e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
@ -546,6 +575,48 @@ func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
continue continue
} }
} }
if claimed.Release {
// Only reached when this node owns the forward (or it is already gone
// and we are the fallback consumer — see the selection loop).
owner := e.state.TopologyOwner(claimed)
if owner != e.ID {
// The forward vanished before we got here; nothing to give up.
continue
}
// Stop our worker. Deliberately NOT Handler.Revoke: that marks the
// link stopped, and the forward is about to run elsewhere — a
// "stopped" flag would make every node treat the re-published task as
// a duplicate and refuse to claim it.
if e.Handler != nil {
if err := e.Handler.Release(ctx, claimed); err != nil {
log.Printf("ring[%s] release %s: %v", e.ID, claimed.ID, err)
}
}
// Drop the entry so the forward has no owner. This is what makes the
// load actually fall (ForwardsOwnedBy counts entries) and what lets a
// fresh creation task pass the duplicate guard above.
e.RemoveTopologyEntry(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort)
// Hand it back to the ring as an ordinary unclaimed forward. It is
// published by THIS node so it rides the very token we are holding —
// no coordinator, no addressed hand-off.
link := claimed.Link
link.Disabled = false
if t := e.SubmitTask(claimed.Local, claimed.Remote, link); t != nil {
log.Printf("ring[%s] released %s: %s→%s:%d back to the ring as %s",
e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
claimed.Link.RemotePort, t.ID)
} else {
log.Printf("ring[%s] released %s but the re-publish was deduped",
e.ID, claimed.ID)
}
if e.Log != nil {
_, _ = e.Log.Append(e.ID, LogForwardStop, map[string]any{
"taskId": claimed.ID, "local": claimed.Local.Name,
"remote": claimed.Remote.Name, "released": true,
})
}
continue
}
if claimed.Restart { if claimed.Restart {
// Only reached when this node owns the forward (or it is already gone // Only reached when this node owns the forward (or it is already gone
// and we are the fallback consumer — see the selection loop). // and we are the fallback consumer — see the selection loop).
@ -960,6 +1031,26 @@ func (e *Engine) SubmitRestart(local store.Local, remote store.Remote, link stor
return t return t
} }
// ReleaseTask publishes a rebalancing task asking the OWNER of this forward to
// give it up. The task rides the token like any other command; it is not
// addressed to a destination, because the ring's normal lowest-load claim
// decides where the forward goes next.
//
// It goes through AddPending rather than writing PendingTasks directly. That is
// not cosmetic: OnToken rebuilds local-only pending commands from the ids it
// never published, and only the Add* helpers record them — a task injected by
// hand was silently dropped at the next state adoption, so it never reached the
// ring at all.
func (e *Engine) ReleaseTask(local store.Local, remote store.Remote, link store.Link) *Task {
link.Disabled = false
t := e.state.AddPending(local, remote, link)
if t == nil {
return nil
}
t.Release = true
return t
}
// HasActiveTask reports whether a forward with this key is genuinely in flight // HasActiveTask reports whether a forward with this key is genuinely in flight
// or running: a non-revocation pending task, or a topology entry that is not // or running: a non-revocation pending task, or a topology entry that is not
// marked disabled. // marked disabled.

View File

@ -12,6 +12,14 @@ type fakeHandler struct {
claim func(ctx context.Context, tk *Task) error claim func(ctx context.Context, tk *Task) error
revoke func(ctx context.Context, tk *Task) error revoke func(ctx context.Context, tk *Task) error
restart func(ctx context.Context, tk *Task) error restart func(ctx context.Context, tk *Task) error
release func(ctx context.Context, tk *Task) error
}
func (h *fakeHandler) Release(ctx context.Context, tk *Task) error {
if h.release != nil {
return h.release(ctx, tk)
}
return nil
} }
func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error { func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error {

View File

@ -21,6 +21,12 @@ type AppHandler struct {
// stopped flag + bring the worker back). Required because a stop keeps the // stopped flag + bring the worker back). Required because a stop keeps the
// topology entry, so the normal claim path will not re-create it. // topology entry, so the normal claim path will not re-create it.
RestartFn func(ctx context.Context, tk *Task) error RestartFn func(ctx context.Context, tk *Task) error
// ReleaseFn stops this node's worker for a forward being handed back to the
// ring, leaving the link ENABLED. When nil, Release falls back to RevokeFn —
// which also stops the worker, but additionally marks the link stopped; that
// is the pre-existing stop semantics and is wrong for a rebalance, so the app
// should provide this.
ReleaseFn func(ctx context.Context, tk *Task) error
} }
// RuntimeLoad implements Handler. // RuntimeLoad implements Handler.
@ -48,6 +54,22 @@ func (a *AppHandler) Revoke(ctx context.Context, tk *Task) error {
return a.RevokeFn(ctx, tk) return a.RevokeFn(ctx, tk)
} }
// Release implements Handler.
func (a *AppHandler) Release(ctx context.Context, tk *Task) error {
if a.ReleaseFn != nil {
return a.ReleaseFn(ctx, tk)
}
// Fall back to the stop path so a host that has not provided ReleaseFn still
// stops the worker (the forward keeps running otherwise). The link ends up
// flagged stopped, which the claim path may then refuse — the host is warned
// by the ring's own "re-publish was deduped" log line rather than silently
// leaving two workers up.
if a.RevokeFn != nil {
return a.RevokeFn(ctx, tk)
}
return nil
}
// Restart implements Handler. // Restart implements Handler.
func (a *AppHandler) Restart(ctx context.Context, tk *Task) error { func (a *AppHandler) Restart(ctx context.Context, tk *Task) error {
if a.RestartFn == nil { if a.RestartFn == nil {

View File

@ -48,6 +48,8 @@ type Handler struct {
Cluster *cluster.Registry Cluster *cluster.Registry
// Ring is the token-ring engine (M6). Nil disables ring routes. // Ring is the token-ring engine (M6). Nil disables ring routes.
Ring *cluster.Engine Ring *cluster.Engine
// Balancer is optional: when nil the balance endpoint reports 404.
Balancer *cluster.Balancer
// SelfAddr is this node's reachable listen address (from -addr), used for // SelfAddr is this node's reachable listen address (from -addr), used for
// cluster discovery so peers can reach back for binary exchange. // cluster discovery so peers can reach back for binary exchange.
SelfAddr string SelfAddr string
@ -114,6 +116,8 @@ func NewServeMux(h *Handler) (http.Handler, error) {
mux.HandleFunc(apiPrefix+"/cluster/nodes", h.auth("read")(h.handleClusterNodes)) mux.HandleFunc(apiPrefix+"/cluster/nodes", h.auth("read")(h.handleClusterNodes))
mux.HandleFunc(apiPrefix+"/cluster/cache", h.auth("read")(h.handleClusterCache)) // GET read; POST write-checked inside mux.HandleFunc(apiPrefix+"/cluster/cache", h.auth("read")(h.handleClusterCache)) // GET read; POST write-checked inside
mux.HandleFunc(apiPrefix+"/cluster/ring", h.auth("read")(h.handleClusterRing)) mux.HandleFunc(apiPrefix+"/cluster/ring", h.auth("read")(h.handleClusterRing))
mux.HandleFunc(apiPrefix+"/cluster/balance", h.auth("read")(h.handleClusterBalance))
mux.HandleFunc(apiPrefix+"/cluster/balance/mode", h.auth("write")(h.handleClusterBalanceMode))
mux.HandleFunc(apiPrefix+"/node/logs", h.auth("read")(h.handleNodeLogs)) mux.HandleFunc(apiPrefix+"/node/logs", h.auth("read")(h.handleNodeLogs))
mux.HandleFunc(apiPrefix+"/cluster/logs/export", h.auth("read")(h.handleClusterLogsExport)) mux.HandleFunc(apiPrefix+"/cluster/logs/export", h.auth("read")(h.handleClusterLogsExport))
@ -634,6 +638,64 @@ func (h *Handler) handleClusterRing(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, h.Ring.Snapshot()) writeJSON(w, http.StatusOK, h.Ring.Snapshot())
} }
// handleClusterBalance reports the load balancer's read-only view: per-node
// scores, whether anything is over the threshold, and the migration it would
// perform. Publishing nothing is the point — an operator asking "is this cluster
// balanced?" currently gets silence from the log in the healthy case, which is
// indistinguishable from balancing being switched off.
func (h *Handler) handleClusterBalance(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
methodNotAllowed(w)
return
}
if h.Ring == nil {
http.Error(w, "ring engine not enabled", http.StatusNotFound)
return
}
if h.Balancer == nil {
http.Error(w, "load balancer not enabled", http.StatusNotFound)
return
}
rep := h.Balancer.Report()
// CanControl is decided by the caller's own level rather than by the
// presence of a stored value, so a read-only principal is never shown a
// switch that would 403.
rep.CanControl = hasLevel(r, "write")
writeJSON(w, http.StatusOK, rep)
}
// handleClusterBalanceMode switches load balancing on/off at runtime.
//
// It is per-node on purpose. Every member decides about its OWN forwards, so the
// switch belongs to each node: an operator may well want to turn releasing on
// everywhere, but the server should not pretend those are one atomic action.
// (A cluster-wide control can be layered on top later; the primitive here is the
// one that actually has an owner.)
//
// Write-level auth: releasing stops a worker and drops a live connection.
func (h *Handler) handleClusterBalanceMode(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
methodNotAllowed(w)
return
}
if h.Balancer == nil {
http.Error(w, "load balancer not enabled", http.StatusNotFound)
return
}
var req struct {
Mode string `json:"mode"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "parse json: "+err.Error(), http.StatusBadRequest)
return
}
if err := h.Balancer.SetModeFromUI(req.Mode); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
writeJSON(w, http.StatusOK, h.Balancer.Report())
}
// handleClusterJoin accepts a newcomer join request: the target node inserts // handleClusterJoin accepts a newcomer join request: the target node inserts
// the newcomer after itself in the ring and returns the updated ring state // the newcomer after itself in the ring and returns the updated ring state
// for the newcomer to adopt. // for the newcomer to adopt.

View File

@ -128,6 +128,11 @@ type Settings struct {
// key). Cleared on explicit detach (detachAsStandalone) so a node that // key). Cleared on explicit detach (detachAsStandalone) so a node that
// intentionally left does NOT auto-rejoin. // intentionally left does NOT auto-rejoin.
ClusterPeers string `json:"clusterPeers,omitempty"` ClusterPeers string `json:"clusterPeers,omitempty"`
// BalanceMode is the load-balancing mode chosen in the UI: "", "off",
// "dry-run" or "on". Empty means "never configured here" so the environment
// variable stays the bootstrap default; a non-empty value wins over it,
// otherwise flipping the switch in the UI would silently revert on restart.
BalanceMode string `json:"balanceMode,omitempty"`
} }
// Forward is a rendered link row attached to a remote. // Forward is a rendered link row attached to a remote.
@ -748,6 +753,13 @@ func (s *Store) Settings() (Settings, error) {
if err := json.Unmarshal([]byte(raw), &st); err != nil { if err := json.Unmarshal([]byte(raw), &st); err != nil {
return Settings{AutoStartProfiles: true, RestartOnExit: true, RestartIntervalSeconds: 5}, nil return Settings{AutoStartProfiles: true, RestartOnExit: true, RestartIntervalSeconds: 5}, nil
} }
// BalanceMode lives in its own column (see UpdateSettings). Surface it here
// too so callers reading Settings get the effective value, but never let it
// overwrite the column on a write-back: UpdateSettings only mirrors a
// non-empty value, so an unrelated settings PUT cannot clear a UI choice.
if mode, ok, err := s.BalanceMode(); err == nil && ok {
st.BalanceMode = mode
}
return st, nil return st, nil
} }
@ -760,7 +772,36 @@ func (s *Store) UpdateSettings(st Settings) error {
"INSERT INTO settings(key, value) VALUES('settings', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", "INSERT INTO settings(key, value) VALUES('settings', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
string(raw), string(raw),
) )
return err if err != nil {
return err
}
if mode := st.BalanceMode; mode != "" {
// Mirror into its own column for the same reason the other flagged
// fields have one: the settings JSON is rewritten wholesale, so a
// clustered field that rides that JSON would flip back and forth between
// nodes instead of converging.
if _, err := s.db.Exec(
"INSERT INTO settings(key, value) VALUES('balance_mode', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
mode,
); err != nil {
return err
}
}
return nil
}
// BalanceMode returns the stored load-balancing mode, and whether one was ever
// stored (false = never configured here; the environment default applies).
func (s *Store) BalanceMode() (string, bool, error) {
var v string
err := s.db.QueryRow("SELECT value FROM settings WHERE key = 'balance_mode'").Scan(&v)
if errors.Is(err, sql.ErrNoRows) {
return "", false, nil
}
if err != nil {
return "", false, err
}
return v, v != "", nil
} }
// SetNodeKey persists just the cluster admission key, preserving all other // SetNodeKey persists just the cluster admission key, preserving all other

View File

@ -2,11 +2,11 @@
// //
// Value is overridable at build time so release artifacts carry the real tag: // Value is overridable at build time so release artifacts carry the real tag:
// //
// go build -ldflags="-X webui4frpc/internal/version.Version=0.1.3" ./cmd/webui4frpc // go build -ldflags="-X webui4frpc/internal/version.Version=0.1.4" ./cmd/webui4frpc
// //
// Kept in its own package because both cmd/ and internal/httpapi report it // Kept in its own package because both cmd/ and internal/httpapi report it
// (main can't be imported, and httpapi must not depend on cmd). // (main can't be imported, and httpapi must not depend on cmd).
package version package version
// Version is the running build's semantic version, without a leading "v". // Version is the running build's semantic version, without a leading "v".
var Version = "0.1.3" var Version = "0.1.4"

View File

@ -1,5 +1,5 @@
Name: webui4frpc Name: webui4frpc
Version: %{?_w4f_version}%{!?_w4f_version:0.1.3} Version: %{?_w4f_version}%{!?_w4f_version:0.1.4}
Release: 1%{?dist} Release: 1%{?dist}
Summary: Visual frpc controller with token-ring cluster Summary: Visual frpc controller with token-ring cluster
License: MIT License: MIT
@ -67,6 +67,11 @@ fi
%doc %{_docdir}/webui4frpc/README.md %doc %{_docdir}/webui4frpc/README.md
%changelog %changelog
* Mon Oct 06 2026 JianFeeeee <jianfeeeee@gitcode> - 0.1.4-1
- 负载均衡:改为每节点自主释放(持有数超过平摊份额且 >=2 时自撤销一条丢回环,由最闲节点认领),不再由 leader 统一调度
- 负载均衡:新增自限上界与下限(撤到平摊值即停、低于 2 条不撤),10 分钟冷却,默认不动作(dry-run)
- 集群页:新增负载均衡视图与每节点模式开关(关闭/观察/启用),模式持久化
- 新增只读接口 GET /cluster/balance 与写接口 POST /cluster/balance/mode
* Sun Oct 05 2026 JianFeeeee <jianfeeeee@gitcode> - 0.1.3-1 * Sun Oct 05 2026 JianFeeeee <jianfeeeee@gitcode> - 0.1.3-1
- 窄屏布局:补 box-sizing 重置、接管画布 fit(缩放下限 0.65)、弹窗宽度钳制、统一滚动容器(修复手机端整页无法滑动) - 窄屏布局:补 box-sizing 重置、接管画布 fit(缩放下限 0.65)、弹窗宽度钳制、统一滚动容器(修复手机端整页无法滑动)
- 集群:停用/启动改为经令牌环跨节点传播,撤销不再静默失效 - 集群:停用/启动改为经令牌环跨节点传播,撤销不再静默失效

View File

@ -79,7 +79,7 @@ cd web && npm run build && rm -rf internal/httpapi/dist && cp -r web/dist intern
### Linux (Debian / Ubuntu) ### Linux (Debian / Ubuntu)
```bash ```bash
sudo dpkg -i webui4frpc-0.1.1-linux-amd64.deb # 或 linux-arm64.deb sudo dpkg -i webui4frpc-0.1.4-linux-amd64.deb # 或 linux-arm64.deb
sudo nano /etc/webui4frpc/webui4frpc.env # 改掉默认密码 sudo nano /etc/webui4frpc/webui4frpc.env # 改掉默认密码
sudo systemctl restart webui4frpc sudo systemctl restart webui4frpc
``` ```
@ -87,7 +87,7 @@ sudo systemctl restart webui4frpc
### Linux (RHEL / Fedora / openEuler / 默认 / 龙蜥) ### Linux (RHEL / Fedora / openEuler / 默认 / 龙蜥)
```bash ```bash
sudo rpm -i webui4frpc-0.1.1-linux-x86_64.rpm sudo rpm -i webui4frpc-0.1.4-linux-x86_64.rpm
sudo vi /etc/webui4frpc/webui4frpc.env sudo vi /etc/webui4frpc/webui4frpc.env
sudo systemctl enable --now webui4frpc sudo systemctl enable --now webui4frpc
``` ```
@ -97,7 +97,7 @@ sudo systemctl enable --now webui4frpc
### Windows ### Windows
双击 `webui4frpc-0.1.1-windows-amd64-setup.exe`(ARM 设备用 `windows-arm64-setup.exe`)。 双击 `webui4frpc-0.1.4-windows-amd64-setup.exe`(ARM 设备用 `windows-arm64-setup.exe`)。
安装向导会询问监听地址与管理员凭据,完成后: 安装向导会询问监听地址与管理员凭据,完成后:
- 程序装到 `C:\Program Files\webui4frpc` - 程序装到 `C:\Program Files\webui4frpc`
@ -107,8 +107,8 @@ sudo systemctl enable --now webui4frpc
### macOS ### macOS
```bash ```bash
sudo installer -pkg webui4frpc-0.1.1-darwin-arm64.pkg -target / # Apple Silicon sudo installer -pkg webui4frpc-0.1.4-darwin-arm64.pkg -target / # Apple Silicon
sudo installer -pkg webui4frpc-0.1.1-darwin-amd64.pkg -target / # Intel sudo installer -pkg webui4frpc-0.1.4-darwin-amd64.pkg -target / # Intel
``` ```
或直接双击 `.pkg` 走图形安装器。安装后 launchd 开机自启,配置在 或直接双击 `.pkg` 走图形安装器。安装后 launchd 开机自启,配置在
@ -126,7 +126,7 @@ sudo launchctl kickstart -k system/com.jianfeeeee.webui4frpc
不想装服务只想跑一下,用对应的 `.tar.gz` / `.zip`,解压即用: 不想装服务只想跑一下,用对应的 `.tar.gz` / `.zip`,解压即用:
```bash ```bash
tar xzf webui4frpc-0.1.1-linux-amd64.tar.gz tar xzf webui4frpc-0.1.4-linux-amd64.tar.gz
./webui4frpc -addr 127.0.0.1:7500 -user admin -password 你的密码 -workdir ./data ./webui4frpc -addr 127.0.0.1:7500 -user admin -password 你的密码 -workdir ./data
``` ```

View File

@ -12,7 +12,7 @@
# 在 release.sh 流程内被调用;依赖 dist/artifacts/<version>/bin/<os>-<arch>/ 下的二进制。 # 在 release.sh 流程内被调用;依赖 dist/artifacts/<version>/bin/<os>-<arch>/ 下的二进制。
set -euo pipefail set -euo pipefail
VERSION="${1:-0.1.3}" VERSION="${1:-0.1.4}"
VERSION="${VERSION#v}" VERSION="${VERSION#v}"
ROOT="$(cd "$(dirname "$0")/.." && pwd)" ROOT="$(cd "$(dirname "$0")/.." && pwd)"
OUT="$ROOT/dist/artifacts/$VERSION" OUT="$ROOT/dist/artifacts/$VERSION"

View File

@ -3,7 +3,7 @@
# webui4frpc 多平台发布脚本 # webui4frpc 多平台发布脚本
# #
# 用法: # 用法:
# ./scripts/release.sh [version] # 默认 version=0.1.3 # ./scripts/release.sh [version] # 默认 version=0.1.4
# #
# 行为: # 行为:
# 1. 构建前端 (web/dist) 并同步到 internal/httpapi/dist (go:embed 用) # 1. 构建前端 (web/dist) 并同步到 internal/httpapi/dist (go:embed 用)
@ -19,7 +19,7 @@
# NO_BUILD_FRONTEND 设为 1 跳过前端构建 (用现有 dist) # NO_BUILD_FRONTEND 设为 1 跳过前端构建 (用现有 dist)
set -euo pipefail set -euo pipefail
VERSION="${1:-0.1.3}" VERSION="${1:-0.1.4}"
VERSION="${VERSION#v}" # 去掉可能的前导 v VERSION="${VERSION#v}" # 去掉可能的前导 v
ROOT="$(cd "$(dirname "$0")/.." && pwd)" ROOT="$(cd "$(dirname "$0")/.." && pwd)"
OUT="$ROOT/dist/artifacts/$VERSION" OUT="$ROOT/dist/artifacts/$VERSION"

View File

@ -2,6 +2,7 @@
import type { import type {
ApiKey, ApiKey,
ApiKeyCreated, ApiKeyCreated,
BalanceReport,
BinaryStatus, BinaryStatus,
CacheResp, CacheResp,
CanvasData, CanvasData,
@ -159,6 +160,21 @@ export const api = {
// M6 token ring snapshot. // M6 token ring snapshot.
ring: () => request<RingSnapshot>("/api/manager/cluster/ring"), ring: () => request<RingSnapshot>("/api/manager/cluster/ring"),
// Load balancer read-only view. Reports per-node scores, whether anything is
// over the threshold, and the migration it would perform. Never mutates the
// cluster. 404 when the balancer is not wired up on that node.
balance: () => request<BalanceReport>("/api/manager/cluster/balance"),
// setBalanceMode switches load balancing on THIS node (off | dry-run | on).
// Per-node by design: each member releases only its own forwards, so there is
// no cluster-wide switch to flip. Requires write level.
setBalanceMode: (mode: string) =>
request<BalanceReport>("/api/manager/cluster/balance/mode", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ mode }),
}),
// M6 cluster lifecycle (创建/加入/退出/移除). // M6 cluster lifecycle (创建/加入/退出/移除).
// clusterCreate: reseed THIS node as a fresh standalone leader (创建集群). // clusterCreate: reseed THIS node as a fresh standalone leader (创建集群).
clusterCreate: () => clusterCreate: () =>

View File

@ -9,332 +9,382 @@ export type CanvasRemote = Remote;
export type CanvasLink = Link; export type CanvasLink = Link;
export interface Local { export interface Local {
name: string; name: string;
ip: string; ip: string;
port: number; port: number;
protocol: string; // tcp | udp | http | https protocol: string; // tcp | udp | http | https
// localOnly: forward stays on this node (loopback kept), no cluster. // localOnly: forward stays on this node (loopback kept), no cluster.
localOnly?: boolean; localOnly?: boolean;
// M1 advanced transport knobs (optional; empty/zero = frpc defaults) // M1 advanced transport knobs (optional; empty/zero = frpc defaults)
useEncryption?: boolean; useEncryption?: boolean;
useCompression?: boolean; useCompression?: boolean;
bandwidthLimit?: string; bandwidthLimit?: string;
poolCount?: number; poolCount?: number;
metadatas?: Record<string, string>; metadatas?: Record<string, string>;
annotations?: Record<string, string>; annotations?: Record<string, string>;
// M2 HTTP/HTTPS routing (http/https only) // M2 HTTP/HTTPS routing (http/https only)
customDomains?: string; // comma-separated customDomains?: string; // comma-separated
subdomain?: string; subdomain?: string;
locations?: string[]; locations?: string[];
hostHeaderRewrite?: string; hostHeaderRewrite?: string;
httpHeaders?: Record<string, string>; httpHeaders?: Record<string, string>;
basicAuthUser?: string; basicAuthUser?: string;
basicAuthPassword?: string; basicAuthPassword?: string;
// M3 load balancing group + health check // M3 load balancing group + health check
lbGroup?: string; lbGroup?: string;
lbGroupKey?: string; lbGroupKey?: string;
healthCheckType?: string; // "" | tcp | http healthCheckType?: string; // "" | tcp | http
healthCheckPath?: string; healthCheckPath?: string;
healthCheckTimeout?: number; healthCheckTimeout?: number;
healthCheckMaxFailed?: number; healthCheckMaxFailed?: number;
healthCheckInterval?: number; healthCheckInterval?: number;
} }
export interface Remote { export interface Remote {
name: string; name: string;
ip: string; ip: string;
port: number; // frps connect port port: number; // frps connect port
token?: string; token?: string;
url?: string; url?: string;
enabled: boolean; enabled: boolean;
// M1 transport section // M1 transport section
transportProtocol?: string; // tcp | quic | kcp | websocket transportProtocol?: string; // tcp | quic | kcp | websocket
transportTls?: boolean; transportTls?: boolean;
transportPool?: number; transportPool?: number;
transportTlsServerName?: string; transportTlsServerName?: string;
// M2: frps vhostHTTPPort // M2: frps vhostHTTPPort
vhostHttpPort?: number; vhostHttpPort?: number;
// M3: local frpc admin API (webServer) // M3: local frpc admin API (webServer)
adminAddr?: string; adminAddr?: string;
adminPort?: number; adminPort?: number;
adminUser?: string; adminUser?: string;
adminPassword?: string; adminPassword?: string;
} }
export interface Link { export interface Link {
id?: number; id?: number;
local: string; local: string;
remote: string; remote: string;
remotePort: number; remotePort: number;
offsetX?: number; offsetX?: number;
offsetY?: number; offsetY?: number;
// group: management label for one-click group start/stop on the forwards // group: management label for one-click group start/stop on the forwards
// page (unrelated to frps load-balancing lbGroup on Local). // page (unrelated to frps load-balancing lbGroup on Local).
group?: string; group?: string;
// disabled: per-forward stop flag. A disabled forward is omitted from the // disabled: per-forward stop flag. A disabled forward is omitted from the
// frpc config (local) / left out of the ring topology (cluster), so stop is // frpc config (local) / left out of the ring topology (cluster), so stop is
// per-card and survives canvas re-saves. // per-card and survives canvas re-saves.
disabled?: boolean; disabled?: boolean;
} }
export interface CanvasData { export interface CanvasData {
locals: Local[]; locals: Local[];
remotes: Remote[]; remotes: Remote[];
links: Link[]; links: Link[];
} }
export interface Settings { export interface Settings {
autoStartProfiles: boolean; autoStartProfiles: boolean;
restartOnExit: boolean; restartOnExit: boolean;
restartIntervalSeconds: number; restartIntervalSeconds: number;
binaryPath?: string; binaryPath?: string;
} }
export interface ProcessStatus { export interface ProcessStatus {
state: string; state: string;
pid?: number; pid?: number;
startTime?: number; startTime?: number;
restartCount: number; restartCount: number;
exitCode?: number; exitCode?: number;
err?: string; err?: string;
} }
export interface ProxyState { export interface ProxyState {
name: string; name: string;
type: string; type: string;
status: string; // new | wait start | start error | running | check failed | closed status: string; // new | wait start | start error | running | check failed | closed
err?: string; err?: string;
local_addr?: string; local_addr?: string;
remote_addr?: string; remote_addr?: string;
} }
export interface StatusProfile { export interface StatusProfile {
name: string; name: string;
enabled: boolean; enabled: boolean;
process: ProcessStatus; process: ProcessStatus;
hasProcess: boolean; hasProcess: boolean;
forwards: { service: string; remotePort: number; localPort?: number }[]; forwards: { service: string; remotePort: number; localPort?: number }[];
// connState: derived frps connection state (we do not control remote frps) // connState: derived frps connection state (we do not control remote frps)
connState?: string; // connected | connecting | failed | not_started | disabled connState?: string; // connected | connecting | failed | not_started | disabled
// M3: admin API state // M3: admin API state
adminEnabled?: boolean; adminEnabled?: boolean;
proxyStates?: ProxyState[]; proxyStates?: ProxyState[];
groupCount?: number; groupCount?: number;
} }
export interface StatusResp { export interface StatusResp {
version: string; version: string;
workDir: string; workDir: string;
settings: Settings; settings: Settings;
services: Local[]; services: Local[];
remotes: Remote[]; remotes: Remote[];
binaryPath: string; binaryPath: string;
profiles: StatusProfile[]; profiles: StatusProfile[];
localStatus: LocalStatus[]; localStatus: LocalStatus[];
// forwards: the forward-centric list rendered as cards on the status page. // forwards: the forward-centric list rendered as cards on the status page.
forwards: ForwardStatus[]; forwards: ForwardStatus[];
// selfId: this node's ring ID (== SelfAddr), to tell whether a cluster // selfId: this node's ring ID (== SelfAddr), to tell whether a cluster
// forward's owner is this node (worker runs here) or a peer. // forward's owner is this node (worker runs here) or a peer.
selfId: string; selfId: string;
} }
// ForwardStatus is one forward (link) in the forward-centric status view. // ForwardStatus is one forward (link) in the forward-centric status view.
// kind distinguishes 本地转发 (localOnly) from 远程转发 (cluster-distributed). // kind distinguishes 本地转发 (localOnly) from 远程转发 (cluster-distributed).
export interface ForwardStatus { export interface ForwardStatus {
local: string; local: string;
remote: string; remote: string;
remotePort: number; remotePort: number;
localOnly: boolean; localOnly: boolean;
kind: "local" | "remote"; kind: "local" | "remote";
ownerId?: string; // remote = topology owner; local = selfId ownerId?: string; // remote = topology owner; local = selfId
active: boolean; // remote = in topology; local = local worker running active: boolean; // remote = in topology; local = local worker running
disabled: boolean; // per-forward stop flag (true = stopped) disabled: boolean; // per-forward stop flag (true = stopped)
group?: string; group?: string;
localIp?: string; localIp?: string;
localPort?: number; localPort?: number;
localProto?: string; localProto?: string;
} }
export interface LocalTargetStatus { export interface LocalTargetStatus {
remote: string; remote: string;
remotePort: number; remotePort: number;
workerState: string; workerState: string;
} }
export interface LocalStatus { export interface LocalStatus {
local: Local; local: Local;
targets: LocalTargetStatus[]; targets: LocalTargetStatus[];
} }
export interface BinaryStatus { export interface BinaryStatus {
binaryPath: string; binaryPath: string;
version?: string; version?: string;
} }
export interface InstallResult { export interface InstallResult {
path: string; path: string;
version: string; version: string;
} }
// M6: cluster node registry + binary cache. // M6: cluster node registry + binary cache.
export interface ClusterNode { export interface ClusterNode {
addr: string; addr: string;
cache?: string[]; cache?: string[];
version?: string; version?: string;
} }
export interface CacheEntry { export interface CacheEntry {
version: string; version: string;
path: string; path: string;
size: number; size: number;
modTime: number; modTime: number;
} }
export interface ClusterNodesResp { export interface ClusterNodesResp {
nodes: ClusterNode[]; nodes: ClusterNode[];
} }
export interface CacheResp { export interface CacheResp {
cache: CacheEntry[]; cache: CacheEntry[];
removed?: string[]; removed?: string[];
} }
// M6 token-ring snapshot (GET /cluster/ring). // M6 token-ring snapshot (GET /cluster/ring).
export interface RingNode { export interface RingNode {
id: string; id: string;
addr: string; addr: string;
isLeader?: boolean; isLeader?: boolean;
alive: boolean; alive: boolean;
load: { memPct: number; netPct: number; forwards?: number }; load: { memPct: number; netPct: number; forwards?: number };
version?: string; version?: string;
cache?: string[]; cache?: string[];
lastSeen?: number; lastSeen?: number;
} }
export interface RingTaskInfo { export interface RingTaskInfo {
id: string; id: string;
revoke?: boolean; revoke?: boolean;
removeNode?: string; // node-removal command (target rides the token to self-remove) removeNode?: string; // node-removal command (target rides the token to self-remove)
local: { name: string }; local: { name: string };
remote: { name: string }; remote: { name: string };
link: { remotePort?: number }; link: { remotePort?: number };
} }
export interface RingTopoEntry { export interface RingTopoEntry {
taskId: string; taskId: string;
ownerId: string; ownerId: string;
active: boolean; active: boolean;
local: { name: string }; local: { name: string };
remote: { name: string }; remote: { name: string };
link: { remotePort?: number }; link: { remotePort?: number };
} }
export interface RingLogEntry { export interface RingLogEntry {
seq: number; seq: number;
node: string; node: string;
kind: string; kind: string;
at?: number; at?: number;
data?: Record<string, any> | null; data?: Record<string, any> | null;
} }
export interface RingSnapshot { export interface RingSnapshot {
selfId: string; selfId: string;
leaderId: string; // 当前 leader(持首令牌 + 判丢失 + 被上家监控) leaderId: string; // 当前 leader(持首令牌 + 判丢失 + 被上家监控)
cycle: number; // 周期号(leader 每收令牌回 +1) cycle: number; // 周期号(leader 每收令牌回 +1)
lastSync: number; // 上次同步时间 (unix sec) lastSync: number; // 上次同步时间 (unix sec)
roundDelayMs?: number; roundDelayMs?: number;
nodes: RingNode[]; nodes: RingNode[];
pending: RingTaskInfo[]; pending: RingTaskInfo[];
topology: RingTopoEntry[]; topology: RingTopoEntry[];
log: RingLogEntry[]; log: RingLogEntry[];
// nodeKey: this node's cluster admission key. Newcomers must present it to // nodeKey: this node's cluster admission key. Newcomers must present it to
// join via this node (handleClusterJoin verifies). Shown on the cluster page // join via this node (handleClusterJoin verifies). Shown on the cluster page
// for the operator to copy; omitted for non-cluster (standalone) nodes. // for the operator to copy; omitted for non-cluster (standalone) nodes.
nodeKey?: string; nodeKey?: string;
}
// BalanceReport is one node's own view of rebalancing (GET /cluster/balance).
//
// It is per-node by design: every member runs the same rule against its own
// forwards, and the answer that matters is the one on the node doing (or
// refusing to do) the releasing. So this reports "my share" first and the peers
// second, and there is no cluster-wide "would migrate" plan — the ring's normal
// lowest-load claim decides where a released forward lands.
export interface BalanceReleasePlan {
local: string;
remote: string;
remotePort: number;
owner: string;
}
export interface BalanceReport {
mode: "off" | "dry-run" | "on";
modeSource: "env" | "stored"; // env = W4F_BALANCE_MODE default, stored = set here
enabled: boolean; // mode !== off
dryRun: boolean; // mode !== on
canControl: boolean; // this principal may switch it (write level)
isMember: boolean;
self: string;
selfForwards: number;
share: number; // the even split: ceil(total / alive)
totalForwards: number;
aliveNodes: number;
selfOverShare: boolean;
// wouldRelease is present whenever this node is over its share — including
// while it is deliberately holding back, which is what blockedBy explains.
wouldRelease?: BalanceReleasePlan;
releasing?: BalanceReleasePlan;
blockedBy?: string; // single-node | below-min | balanced | nothing-active | cooldown | in-flight
cooldownSec: number;
lastReleaseAt?: number;
nodes: BalanceNodeReport[];
}
export interface BalanceNodeReport {
id: string;
alive: boolean;
isLeader: boolean;
forwards: number;
memPct: number;
netPct: number;
score: number;
} }
// ---- Auth / accounts / API keys (M7) ---- // ---- Auth / accounts / API keys (M7) ----
export type AccessLevel = 'read' | 'write' | 'admin' | 'superadmin'; export type AccessLevel = "read" | "write" | "admin" | "superadmin";
// MeResp is the authenticated principal (GET /me). The frontend uses level to // MeResp is the authenticated principal (GET /me). The frontend uses level to
// gate the UI: viewer (= auditors) sees read + export controls only. // gate the UI: viewer (= auditors) sees read + export controls only.
export interface MeResp { export interface MeResp {
type: 'user' | 'service' | 'key'; type: "user" | "service" | "key";
name: string; name: string;
level: AccessLevel; level: AccessLevel;
userId?: number; userId?: number;
} }
export interface User { export interface User {
id: number; id: number;
username: string; username: string;
role: 'admin' | 'viewer' | 'superadmin'; role: "admin" | "viewer" | "superadmin";
enabled: boolean; enabled: boolean;
system?: boolean; // flag-synced built-in account (UI read-only) system?: boolean; // flag-synced built-in account (UI read-only)
createdAt: number; createdAt: number;
lastLoginAt: number; lastLoginAt: number;
} }
export interface ApiKey { export interface ApiKey {
id: number; id: number;
userId: number; userId: number;
prefix: string; prefix: string;
label: string; label: string;
scope: AccessLevel; scope: AccessLevel;
createdAt: number; createdAt: number;
lastUsedAt: number; lastUsedAt: number;
expiresAt?: number; expiresAt?: number;
} }
// ApiKeyCreated is the one-time response from POST /apikeys: the plaintext key // ApiKeyCreated is the one-time response from POST /apikeys: the plaintext key
// is shown exactly once and never retrievable again. // is shown exactly once and never retrievable again.
export interface ApiKeyCreated { export interface ApiKeyCreated {
id: number; id: number;
userId: number; userId: number;
label: string; label: string;
scope: AccessLevel; scope: AccessLevel;
prefix: string; prefix: string;
createdAt: number; createdAt: number;
key: string; key: string;
} }
export interface CanvasExportEnvelope { export interface CanvasExportEnvelope {
_type: 'webui4frpc-canvas'; _type: "webui4frpc-canvas";
version: number; version: number;
exportedAt: number; exportedAt: number;
exporter: string; exporter: string;
canvas: CanvasData; canvas: CanvasData;
} }
// Worker-log bundle (GET /cluster/logs/export): one entry per alive ring node. // Worker-log bundle (GET /cluster/logs/export): one entry per alive ring node.
export interface WorkerLogEntry { export interface WorkerLogEntry {
name: string; name: string;
remote: string; remote: string;
state: string; state: string;
lines: string; lines: string;
} }
export interface NodeLogsResp { export interface NodeLogsResp {
node: string; node: string;
workers: WorkerLogEntry[]; workers: WorkerLogEntry[];
} }
export interface ClusterWorkerLogNode { export interface ClusterWorkerLogNode {
id: string; id: string;
addr: string; addr: string;
ok: boolean; ok: boolean;
error?: string; error?: string;
logs?: NodeLogsResp; logs?: NodeLogsResp;
} }
export interface WorkerLogBundle { export interface WorkerLogBundle {
_type: 'webui4frpc-worker-logs'; _type: "webui4frpc-worker-logs";
version: number; version: number;
exportedAt: number; exportedAt: number;
requester: string; requester: string;
nodes: ClusterWorkerLogNode[]; nodes: ClusterWorkerLogNode[];
} }

View File

@ -121,6 +121,131 @@
</div> </div>
</section> </section>
<!-- Load balancer (read-only). Shown only when the endpoint responds: a
404 means this node has no balancer wired up, and an absent section is
the honest signal for "this build doesn't do balancing". -->
<section class="section w4f-card bal-section" v-if="bal">
<h3 class="section-title">
<span class="k-ic">⚖</span>负载均衡
<!-- The switch edits the mode ON THIS NODE. Every member decides about
its own forwards, so a cluster-wide toggle would be a lie: there is
no single action that flips them together. -->
<el-select
v-if="bal.canControl"
v-model="balModeChoice"
size="small"
class="bal-mode"
:disabled="modeBusy"
@change="onModeChange"
>
<el-option label="关闭" value="off" />
<el-option label="观察模式(不释放)" value="dry-run" />
<el-option label="启用(会释放)" value="on" />
</el-select>
<span class="w4f-tag" v-else v-text="'只读'" ></span>
<span class="bal-src" v-if="bal.modeSource === 'stored'">已保存</span>
<span class="bal-src" v-else>来自启动参数</span>
<span class="w4f-tag" v-if="!bal.isMember" v-text="'非集群成员'"></span>
<span class="w4f-tag" v-else-if="bal.releasing" v-text="'释放进行中'"></span>
<span class="w4f-tag w4f-tag-warn" v-else-if="bal.selfOverShare" v-text="'本机超出份额'"></span>
<span class="w4f-tag w4f-tag-ok" v-else v-text="'本机在份额内'"></span>
<span class="title-actions">
<button class="w4f-btn ghost small" :disabled="busy" @click="load">⟳ 刷新</button>
</span>
</h3>
<div class="bal-thresholds">
<span>
规则:本机转发 <b>{{ bal.selfForwards }}</b> 条
&gt; 平摊份额 <b>{{ bal.share }}</b>
且至少 {{ 2 }} 条时,本机自行释放一条
</span>
<span>集群共 {{ bal.totalForwards }} 条 / {{ bal.aliveNodes }} 台存活</span>
<span>两次释放间隔 ≥ {{ Math.round(bal.cooldownSec / 60) }} 分钟</span>
<span v-if="bal.dryRun" class="bal-hint">
设 <code>W4F_BALANCE_MODE=on</code> 才会真正释放
</span>
</div>
<p class="bal-note">
每个节点各自判断自己的转发,超了就自己撤销一条并丢回环上,
由环节点按「最闲者优先」重新认领。没有中心调度。
</p>
<!-- Per-node forward counts. Bars are scaled to the busiest node so the
shape reads whether the cluster holds 3 forwards or 300; the share
marker is what actually decides, not the bar. -->
<div class="bal-nodes">
<div
v-for="n in balNodes"
:key="n.id"
class="bal-node"
:class="{ self: n.id === bal.self, over: n.forwards > bal.share }"
>
<div class="bn-top">
<span class="bn-id">{{ shortId(n.id) }}</span>
<span v-if="n.id === bal.self" class="w4f-tag w4f-tag-primary">本机</span>
<span v-if="n.isLeader" class="w4f-tag w4f-tag-info">LEADER</span>
<span v-if="!n.alive" class="w4f-tag w4f-tag-danger">离线</span>
<span v-if="n.alive && n.forwards > bal.share" class="w4f-tag w4f-tag-warn">超出份额</span>
</div>
<div class="w4f-bar" :class="{ err: n.alive && n.forwards > bal.share }">
<i :style="{ width: barWidth(n.forwards) + '%' }" />
</div>
<div class="bn-stats">
<span>转发 <b>{{ n.forwards }}</b></span>
<span>内存 <b>{{ n.memPct.toFixed(1) }}%</b></span>
<span>网络 <b>{{ n.netPct.toFixed(1) }}%</b></span>
<span class="bn-score">份额 <b>{{ bal.share }}</b></span>
</div>
</div>
<div v-if="!balNodes.length" class="empty">暂无节点负载数据</div>
</div>
<!-- What THIS node would do. The answer is per-node by design: every
member runs the same rule on its own forwards, so there is no
cluster-wide plan and no "the leader will move X to Y". -->
<div class="bal-plan">
<div v-if="bal.releasing" class="bp-inflight">
<span class="w4f-tag w4f-tag-warn">释放进行中</span>
<span class="bp-text">
本机已释放 {{ bal.releasing.local }}:{{ bal.releasing.remotePort }},
等待环把它认领到别的节点
</span>
</div>
<div v-else-if="bal.wouldRelease && !bal.blockedBy" class="bp-would">
<span class="w4f-tag w4f-tag-warn">将释放</span>
<span class="bp-text">
本机将撤销 {{ bal.wouldRelease.local }}:{{ bal.wouldRelease.remotePort }}
并丢回环上,由最闲节点接管
</span>
</div>
<div v-else-if="bal.wouldRelease && bal.blockedBy === 'cooldown'" class="bp-waiting">
<span class="w4f-tag w4f-tag-info">冷却中</span>
<span class="bp-text">
本机超出份额,但距上次释放不足
{{ Math.round(bal.cooldownSec / 60) }} 分钟
<template v-if="bal.lastReleaseAt">
(已过 {{ cooldownAgo }})
</template>
</span>
</div>
<div v-else-if="bal.wouldRelease" class="bp-waiting">
<span class="w4f-tag w4f-tag-info">暂不释放</span>
<span class="bp-text">本机超出份额,原因:{{ blockedLabel }}</span>
</div>
<div v-else-if="!bal.isMember" class="bp-ok">
<span class="w4f-tag w4f-tag-ok">单节点</span>
<span class="bp-text">本机未加入集群,无需均衡</span>
</div>
<div v-else class="bp-ok">
<span class="w4f-tag w4f-tag-ok">已均衡</span>
<span class="bp-text">
本机持有 {{ bal.selfForwards }} 条,未超平摊份额 {{ bal.share }},无需释放
</span>
</div>
</div>
</section>
<!-- Active forward topology --> <!-- Active forward topology -->
<section class="section w4f-card"> <section class="section w4f-card">
<h3 class="section-title"><span class="k-ic">→</span>活跃转发拓扑</h3> <h3 class="section-title"><span class="k-ic">→</span>活跃转发拓扑</h3>
@ -199,14 +324,18 @@
</template> </template>
<script setup lang="ts"> <script setup lang="ts">
import { ref, reactive, computed, onMounted, onBeforeUnmount } from 'vue' import { ref, reactive, computed, onMounted, onBeforeUnmount, watch } from 'vue'
import { ElMessage } from 'element-plus' import { ElMessage } from 'element-plus'
import { api, downloadAuditCsv } from '../api' import { api, downloadAuditCsv } from '../api'
import { canWrite } from '../auth' import { canWrite } from '../auth'
import type { RingSnapshot, RingTaskInfo, RingLogEntry } from '../types' import type { BalanceReport, RingSnapshot, RingTaskInfo, RingLogEntry } from '../types'
const ring = ref<RingSnapshot | null>(null) const ring = ref<RingSnapshot | null>(null)
const busy = ref(false) const busy = ref(false)
// bal is null when the node has no balancer wired up (endpoint 404s). The
// section is simply not rendered then, which is the honest signal for "this
// build does not do balancing" — better than an empty panel implying it does.
const bal = ref<BalanceReport | null>(null)
let timer: number | null = null let timer: number | null = null
let visHandler: (() => void) | null = null let visHandler: (() => void) | null = null
@ -214,8 +343,76 @@ const joinDialog = reactive({ visible: false, addr: '', key: '' })
const load = async () => { const load = async () => {
try { ring.value = await api.ring() } catch { /* keep last */ } try { ring.value = await api.ring() } catch { /* keep last */ }
// Loaded alongside the ring, but unlike it a failure clears the view: the
// balancer report is a decision surface, and showing a stale "已均衡" after the
// endpoint starts failing would be worse than showing nothing.
try { bal.value = await api.balance() } catch { bal.value = null }
} }
// ---- load balancer mode switch ---- //
// The select is bound to a local mirror so the UI can show the pending choice
// while the request is in flight, and snap back if the server rejects it.
const balModeChoice = ref<string>('dry-run')
const modeBusy = ref(false)
watch(
() => bal.value?.mode,
(m) => { if (m) balModeChoice.value = m },
{ immediate: true },
)
const onModeChange = async (m: string) => {
if (!bal.value || modeBusy.value) return
modeBusy.value = true
try {
bal.value = await api.setBalanceMode(m)
ElMessage.success(
m === 'on' ? '本机已启用负载释放' : m === 'off' ? '本机已关闭负载均衡' : '本机已切到观察模式',
)
} catch (e) {
// Snap back: leaving the select showing a mode the server did not accept
// would misrepresent what the node is doing.
balModeChoice.value = bal.value.mode
ElMessage.error('切换失败:' + (e instanceof Error ? e.message : String(e)))
} finally {
modeBusy.value = false
}
}
// ---- load balancer view ---- //
// Each node decides about its OWN forwards, so this page has no cluster-wide
// plan to show — only what THIS node holds and what it is about to do.
// Ordered by descending count: the busiest node first is what an operator
// scanning for a hotspot wants.
const balNodes = computed(() => [...(bal.value?.nodes ?? [])].sort((a, b) => b.forwards - a.forwards))
// barWidth scales against the busiest node rather than an absolute, so the shape
// is readable whether the cluster holds 3 forwards or 300. The bar is context;
// `share` is what actually decides whether a node is over.
const barWidth = (forwards: number) => {
const max = Math.max(...(bal.value?.nodes ?? []).map((n) => n.forwards), 1)
return Math.min(100, Math.round((forwards / max) * 100))
}
// blockedLabel renders the server's machine reason as something readable. The
// reason matters: "over its share but not acting" is otherwise indistinguishable
// from "broken".
const blockedLabel = computed(() => {
switch (bal.value?.blockedBy) {
case 'cooldown': return '冷却期未过'
case 'in-flight': return '上一次释放尚未确认'
case 'nothing-active': return '本机可释放的转发都已停用'
case 'below-min': return '持有量低于下限'
case 'single-node': return '单节点环'
default: return bal.value?.blockedBy || '未说明'
}
})
// cooldownAgo says how long ago the last release was, so an operator can see
// whether the cooldown is about to expire.
const cooldownAgo = computed(() => {
const t = bal.value?.lastReleaseAt ?? 0
if (!t) return ''
const s = Math.max(0, Math.floor(nowMs.value / 1000) - t)
if (s < 60) return s + ' 秒前'
return Math.floor(s / 60) + ' 分钟前'
})
const shortId = (id: string) => (id.length > 22 ? '…' + id.slice(-20) : id) const shortId = (id: string) => (id.length > 22 ? '…' + id.slice(-20) : id)
// copyNodeKey copies this node's admission key to the clipboard so the operator // copyNodeKey copies this node's admission key to the clipboard so the operator
@ -260,6 +457,8 @@ const stateDotClass = computed(() => {
// ---- last-sync freshness (ops cares "环还活着吗", not the cycle counter) ---- // ---- last-sync freshness (ops cares "环还活着吗", not the cycle counter) ----
const now = ref(Date.now()) const now = ref(Date.now())
// nowMs mirrors `now` as a plain number for the balance section's minute math.
const nowMs = computed(() => now.value)
let tick: number | null = null let tick: number | null = null
const lastSyncAgo = computed(() => { const lastSyncAgo = computed(() => {
const ls = ring.value?.lastSync ?? 0 const ls = ring.value?.lastSync ?? 0
@ -638,4 +837,66 @@ onBeforeUnmount(() => {
.kpis { grid-template-columns: minmax(0, 1fr); } .kpis { grid-template-columns: minmax(0, 1fr); }
.page-title { font-size: 15.5px; } .page-title { font-size: 15.5px; }
} }
/* ---- Load balancer section ---- */
.bal-section .section-title { display: flex; align-items: center; gap: 8px; flex-wrap: wrap; }
.bal-section .section-title .k-ic { margin-right: 0; }
.bal-thresholds {
display: flex; flex-wrap: wrap; gap: 6px 14px;
font-size: 11.5px; color: var(--w4f-muted);
margin-bottom: 10px;
}
.bal-hint { color: var(--w4f-warning); }
.bal-hint code {
font-family: var(--w4f-mono, ui-monospace, monospace);
background: var(--w4f-card-2); padding: 1px 4px; border-radius: 3px;
}
.bal-nodes {
display: grid; gap: 8px;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
}
.bal-node {
border: 1px solid var(--w4f-line);
border-radius: 8px;
padding: 9px 11px;
background: var(--w4f-card-2);
min-width: 0;
}
.bal-node.self { border-color: var(--w4f-primary); }
.bal-node.over { border-color: var(--w4f-warning); }
.bal-mode { width: 168px; }
.bal-src { font-size: 11px; color: var(--w4f-muted); }
.bal-note {
margin: 0 0 10px; font-size: 11.5px; line-height: 1.55;
color: var(--w4f-muted);
}
.bn-top { display: flex; align-items: center; gap: 6px; flex-wrap: wrap; margin-bottom: 7px; }
.bn-id {
font-family: var(--w4f-mono, ui-monospace, monospace);
font-size: 12px; font-weight: 700;
overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
}
.bn-stats {
display: flex; flex-wrap: wrap; gap: 3px 12px;
font-size: 11.5px; color: var(--w4f-muted); margin-top: 7px;
}
.bn-stats b { color: var(--w4f-text); font-variant-numeric: tabular-nums; }
.bn-score { margin-left: auto; }
.bal-plan { margin-top: 11px; }
.bp-inflight, .bp-would, .bp-waiting, .bp-ok {
display: flex; align-items: center; gap: 9px; flex-wrap: wrap;
padding: 8px 11px; border-radius: 8px;
font-size: 12.5px;
background: var(--w4f-card-2);
border: 1px solid var(--w4f-line);
}
.bp-inflight { border-color: var(--w4f-warning); }
.bp-text { min-width: 0; overflow-wrap: anywhere; }
@media (max-width: 380px) {
.bn-score { margin-left: 0; }
}
</style> </style>