From 60597109c4aa99fb92a9330ffc30be9716571274 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 11:20:06 +0800 Subject: [PATCH 01/13] docs: add TRON energy expiry monitoring design spec Co-Authored-By: Claude Fable 5 --- .../2026-08-14-tron-energy-expiry-design.md | 134 ++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 docs/superpowers/specs/2026-08-14-tron-energy-expiry-design.md diff --git a/docs/superpowers/specs/2026-08-14-tron-energy-expiry-design.md b/docs/superpowers/specs/2026-08-14-tron-energy-expiry-design.md new file mode 100644 index 0000000..a2cce30 --- /dev/null +++ b/docs/superpowers/specs/2026-08-14-tron-energy-expiry-design.md @@ -0,0 +1,134 @@ +# TRON Energy 到期监控设计 + +日期：2026-08-14 +需求来源：`tron-energy-expiry-alert-requirements.zh-CN.md`（TRON 地址 Energy 到期监控与报警服务需求文档） + +## 目标 + +监控目标 TRON 地址的入站 Energy 代理，每小时计算"lookahead（默认 72h）后仍受链上锁定保护的 Energy"，低于阈值时经 Slack 报警，支持去重、持续提醒、恢复通知，查询失败时发独立的监控异常报警。 + +## 已确认决策 + +| 决策点 | 结论 | +|---|---| +| 代码位置 | `chains/tron` 内新增独立检查器（方案 B），独立 goroutine + 独立 ticker | +| 链上查询 | 复用现有 gotron-sdk gRPC 连接，不新增 TronGrid HTTP client | +| 持久化 | JSON 状态文件（仿 blockstore 风格），不引入数据库 | +| 报警渠道 | 仅 Slack，复用 `util.Alarm` | + +## 架构 + +``` +Monitor.Sync() + ├── 现有 balance/energy/token 轮询 goroutine（60s，不动） + └── EnergyExpiryChecker goroutine（每个配置了 protectedThreshold 的地址） + ticker（默认 60min） + ↓ + 查询层 con.go：GetInboundDelegatedEnergy + GetAccountResource + ↓ （任一失败 → UNKNOWN，不产出部分数据） + 计算层 energy_calc.go：纯函数，代理列表 → 快照 + ↓ + 状态机 energy_state.go：纯逻辑，快照+上轮状态 → 动作（首报/重提醒/恢复/异常/静默） + ↓ + 持久化 energy_store.go：JSON 原子写 + 报警 util.Alarm（Slack） +``` + +## 配置 + +扩展现有 `config.Energy` 结构（`internal/config/config.go`），向后兼容：`protectedThreshold == 0` 时新监控不启用，原 `waterline` 行为不变。 + +```json +"energy": [{ + "address": "TT6GDYkpHPVk24w9he9pavbagtzqBRS3XP", + "waterline": 100000, + "protectedThreshold": 10000000, + "recoveryThreshold": 10500000, + "lookaheadHours": 72, + "checkIntervalMinutes": 60, + "repeatIntervalHours": 12 +}] +``` + +默认值：`recoveryThreshold` = `protectedThreshold × 1.05`（向上取整）；`lookaheadHours` = 72；`checkIntervalMinutes` = 60；`repeatIntervalHours` = 12。 + +配置热重载（reloader）沿用现有 `UpdateCfg` 机制，checker 每轮 tick 读最新快照配置。 + +## 查询层（chains/tron/con.go 新增） + +`GetInboundDelegatedEnergy(target string) ([]*core.DelegatedResource, error)`： + +1. `GetDelegatedResourceAccountIndexV2(target)` → 读 `FromAccounts`（注意：SDK 现成的 `GetDelegatedResourcesV2` 遍历的是 `ToAccounts`，方向相反，不能用）。 +2. 逐个 `GetDelegatedResourceV2(from → target)`，请求间隔 500ms。 +3. 单请求失败重试 3 次，指数退避（1s/2s/4s）。 +4. 任一步最终失败 → 返回 error，本轮整体 UNKNOWN。禁止以部分结果计算总量。 + +同轮再调 `GetAccountResource(target)` 取 `EnergyLimit`、`EnergyUsed`、`TotalEnergyLimit`、`TotalEnergyWeight`，失败同样 UNKNOWN。 + +## 计算层（chains/tron/energy_calc.go，纯函数） + +输入：代理明细列表、AccountResource、`now`。输出快照： + +- `protectedEnergy24h / 72h / 7d`：到期时间严格晚于 `now+窗口` 的入站代理估算 Energy 之和； +- 排除：已到期、窗口内到期、`expire_time_for_energy == 0`（无锁定，随时可撤回）； +- 换算：`estimatedEnergy = frozen_balance_for_energy / 1e6 × TotalEnergyLimit / TotalEnergyWeight`，每轮用最新全网参数； +- 辅助：当前 EnergyLimit/EnergyUsed/剩余、入站代理总估算 Energy、未来 72h 内到期笔数与 Energy、最近一笔到期时间。 + +`TotalEnergyWeight == 0` 视为数据异常 → UNKNOWN。 + +## 状态机（chains/tron/energy_state.go，纯逻辑） + +三态：`OK` / `ALERT` / `UNKNOWN`。转移与动作： + +| 场景 | 动作 | +|---|---| +| OK→ALERT（protectedEnergy72h < protectedThreshold） | 立即首报 | +| ALERT 持续 | 距上次发送 ≥ repeatInterval 时重提醒 | +| ALERT 持续且缺口比最近一次已发送值扩大 >20% | 立即再报 | +| ALERT→OK（protectedEnergy72h ≥ recoveryThreshold） | 发恢复通知一次 | +| ALERT 且值在 [protectedThreshold, recoveryThreshold) | 维持 ALERT，不发（恢复缓冲） | +| 查询失败 | 转 UNKNOWN，保留上一 OK/ALERT 基线；连续失败 ≥3 轮发一次监控异常报警 | +| UNKNOWN→查询恢复 | 按当前值重新判定，与保留基线比较决定是否首报/恢复 | + +Energy 阈值报警与监控异常报警文案明确区分，互不混用。 + +## 报警文案 + +按需求 6.1 模板，含：地址、预测时间（now+lookahead）、受保护 Energy、阈值、缺口、未来 3 天到期笔数与量、最近到期时间、查询时间。内部计算与存储用 UTC，展示转换为 `Asia/Singapore`（常量，后续需要再做成配置）。发送走 `util.Alarm`。 + +## 持久化（chains/tron/energy_store.go） + +- 每地址一个文件：`/energy_state_
.json`； +- 内容：报警状态（当前态、首报/最近发送/最近恢复时间、最近已发送指标值、连续失败计数）+ 最近 168 轮快照环形缓冲（约 7 天，用于追溯）； +- 写入：临时文件 + `os.Rename` 原子替换； +- 启动时加载：状态存在则恢复，避免重启后重复首报；文件损坏则告警并按全新状态启动。 + +## 错误处理 + +- 429/5xx/超时：指数退避重试（查询层）； +- 部分代理详情缺失、字段异常（如 TotalEnergyWeight=0）：整轮 UNKNOWN； +- 连续 3 轮失败：监控异常报警（独立文案）； +- 绝不把查询失败当作 Energy=0 触发阈值报警。 + +## 测试 + +- `energy_calc_test.go`：到期边界（恰好等于 cutoff 不计入）、无锁定代理排除、已到期排除、换算公式、weight=0 异常； +- `energy_state_test.go`：首报、12h 去重、20% 缺口再报、恢复缓冲（阈值与恢复阈值之间不发）、恢复通知、UNKNOWN 不误报、连续 3 次失败才发异常、UNKNOWN 恢复后基线比较； +- `energy_store_test.go`：读写往返、原子性、损坏文件降级； +- 查询层抽小接口（仅覆盖用到的 3 个方法）供上述测试 mock，gRPC 真实调用不在单测范围。 + +计算层与状态机为纯函数，TDD 先行。 + +## 边界条件（继承需求第 10 章） + +- 同一代理方多笔代理可能被链上聚合，不承诺逐订单展示； +- 链上到期时间是"最早可撤回时间"，非平台承诺撤回时间； +- 换算比例每轮动态更新； +- 时间统一 UTC 存储，展示转 Asia/Singapore。 + +## 明确不做（YAGNI） + +- Telegram / 企业微信渠道； +- SQLite / PostgreSQL； +- Prometheus 指标、HTTP 健康检查、Grafana； +- 出租平台订单 API 对接。 From f8f1977712bc909e9356479118d7a6246334a44e Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 11:41:51 +0800 Subject: [PATCH 02/13] docs: add tron energy expiry implementation plan Co-Authored-By: Claude Fable 5 --- .../plans/2026-08-14-tron-energy-expiry.md | 1469 +++++++++++++++++ 1 file changed, 1469 insertions(+) create mode 100644 docs/superpowers/plans/2026-08-14-tron-energy-expiry.md diff --git a/docs/superpowers/plans/2026-08-14-tron-energy-expiry.md b/docs/superpowers/plans/2026-08-14-tron-energy-expiry.md new file mode 100644 index 0000000..b28c5c3 --- /dev/null +++ b/docs/superpowers/plans/2026-08-14-tron-energy-expiry.md @@ -0,0 +1,1469 @@ +# TRON Energy 到期监控实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 在 chains/tron 中新增独立的 Energy 到期检查器：每小时查询目标地址的入站 Energy 代理，计算 lookahead（默认 72h）后仍受锁定保护的 Energy，低于阈值经 Slack 报警，支持去重/持续提醒/恢复通知/监控异常报警，状态持久化到 JSON 文件。 + +**Architecture:** 查询层（`con.go`，gRPC 入站代理封装）→ 计算层（`energy_calc.go`，纯函数）→ 状态机（`energy_state.go`，纯逻辑）→ 持久化（`energy_store.go`，JSON 原子写）→ 检查器（`energy.go`，独立 goroutine 每分钟 tick，按各地址周期调度）。配置扩展 `config.Energy`，`protectedThreshold==0` 时不启用（向后兼容）。 + +**Tech Stack:** Go、gotron-sdk（gRPC，已有依赖）、log15、util.Alarm（Slack）。无新增第三方依赖。 + +**Spec:** `docs/superpowers/specs/2026-08-14-tron-energy-expiry-design.md` + +**测试约定:** 表驱动测试（仿 `internal/config/waterline_test.go`）。测试命令 `go test ./chains/tron/... ./internal/config/... -v -run `。 + +--- + +### Task 1: 扩展 config.Energy 结构与默认值 + +**Files:** +- Modify: `internal/config/config.go`（Energy 结构体，约第 71 行） +- Test: `internal/config/energy_test.go`（新建） + +- [ ] **Step 1: 写失败的测试** + +创建 `internal/config/energy_test.go`： + +```go +package config + +import "testing" + +func TestEnergyApplyExpiryDefaults(t *testing.T) { + tests := []struct { + name string + in Energy + want Energy + }{ + { + name: "disabled when protectedThreshold is zero", + in: Energy{Address: "T1", Waterline: 100}, + want: Energy{Address: "T1", Waterline: 100}, + }, + { + name: "fills all defaults", + in: Energy{Address: "T1", ProtectedThreshold: 10000000}, + want: Energy{ + Address: "T1", + ProtectedThreshold: 10000000, + RecoveryThreshold: 10500000, // ×1.05 向上取整 + LookaheadHours: 72, + CheckIntervalMinutes: 60, + RepeatIntervalHours: 12, + }, + }, + { + name: "recovery rounds up", + in: Energy{Address: "T1", ProtectedThreshold: 3}, + want: Energy{ + Address: "T1", + ProtectedThreshold: 3, + RecoveryThreshold: 4, // 3*1.05=3.15 → ceil 4 + LookaheadHours: 72, + CheckIntervalMinutes: 60, + RepeatIntervalHours: 12, + }, + }, + { + name: "explicit values are kept", + in: Energy{ + Address: "T1", + ProtectedThreshold: 100, + RecoveryThreshold: 120, + LookaheadHours: 24, + CheckIntervalMinutes: 30, + RepeatIntervalHours: 6, + }, + want: Energy{ + Address: "T1", + ProtectedThreshold: 100, + RecoveryThreshold: 120, + LookaheadHours: 24, + CheckIntervalMinutes: 30, + RepeatIntervalHours: 6, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tt.in + got.ApplyExpiryDefaults() + if got != tt.want { + t.Fatalf("ApplyExpiryDefaults()=%+v, want %+v", got, tt.want) + } + }) + } +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./internal/config/ -run TestEnergyApplyExpiryDefaults -v` +Expected: FAIL（编译错误：Energy 无 ProtectedThreshold 字段 / ApplyExpiryDefaults 未定义） + +- [ ] **Step 3: 实现** + +在 `internal/config/config.go` 中把 Energy 结构体（原第 71-74 行）替换为： + +```go +type Energy struct { + Address string `json:"address"` + Waterline int64 `json:"waterline"` + // Expiry monitoring (Stake 2.0 delegation lock). Disabled when + // ProtectedThreshold <= 0; remaining fields are defaulted by + // ApplyExpiryDefaults. + ProtectedThreshold int64 `json:"protectedThreshold"` + RecoveryThreshold int64 `json:"recoveryThreshold"` + LookaheadHours int64 `json:"lookaheadHours"` + CheckIntervalMinutes int64 `json:"checkIntervalMinutes"` + RepeatIntervalHours int64 `json:"repeatIntervalHours"` +} + +// ApplyExpiryDefaults fills unset expiry-monitoring fields. No-op when the +// feature is disabled (ProtectedThreshold <= 0). +func (e *Energy) ApplyExpiryDefaults() { + if e.ProtectedThreshold <= 0 { + return + } + if e.RecoveryThreshold <= 0 { + // protected × 1.05, rounded up + e.RecoveryThreshold = e.ProtectedThreshold + (e.ProtectedThreshold+19)/20 + } + if e.LookaheadHours <= 0 { + e.LookaheadHours = 72 + } + if e.CheckIntervalMinutes <= 0 { + e.CheckIntervalMinutes = 60 + } + if e.RepeatIntervalHours <= 0 { + e.RepeatIntervalHours = 12 + } +} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `go test ./internal/config/ -run TestEnergyApplyExpiryDefaults -v` +Expected: PASS。再跑 `go test ./internal/config/` 确认没破坏现有测试（注意：仓库里 internal/config 有用户未提交的改动，若原本就有失败先记录，不要归因到本任务）。 + +- [ ] **Step 5: Commit** + +```bash +git add internal/config/config.go internal/config/energy_test.go +git commit -m "feat(config): add tron energy expiry monitoring fields to Energy" +``` + +--- + +### Task 2: 计算层 energy_calc.go（纯函数） + +**Files:** +- Create: `chains/tron/energy_calc.go` +- Test: `chains/tron/energy_calc_test.go` + +- [ ] **Step 1: 写失败的测试** + +创建 `chains/tron/energy_calc_test.go`： + +```go +package tron + +import ( + "testing" + "time" +) + +func TestComputeEnergySnapshot(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + lookahead := 72 * time.Hour + hourMs := int64(3600 * 1000) + // energyPerTRX = TotalEnergyLimit/TotalEnergyWeight = 180e9/6e9 = 30 + res := ResourceParams{ + EnergyLimit: 500_000, + EnergyUsed: 120_000, + TotalEnergyLimit: 180_000_000_000, + TotalEnergyWeight: 6_000_000_000, + } + // 1000 TRX = 1_000_000_000 sun → 30_000 energy + trx1000 := int64(1_000_000_000) + + dels := []DelegationDetail{ + {From: "A", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() + 200*hourMs}, // 8天后到期：全部窗口受保护 + {From: "B", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() + 48*hourMs}, // 48h 后到期：只受 24h 保护 + {From: "C", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() - hourMs}, // 已到期 + {From: "D", FrozenBalanceSun: trx1000, ExpireTimeMs: 0}, // 无锁定 + {From: "E", FrozenBalanceSun: trx1000, ExpireTimeMs: now.Add(lookahead).UnixMilli()}, // 恰好=cutoff：不计入（严格晚于） + } + + snap, err := ComputeEnergySnapshot("T1", dels, res, now, lookahead) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if snap.TotalInboundEnergy != 150_000 { + t.Errorf("TotalInboundEnergy=%d, want 150000", snap.TotalInboundEnergy) + } + if snap.Protected24H != 90_000 { // A + B + E（E 到期于 72h > 24h 窗口） + t.Errorf("Protected24H=%d, want 90000", snap.Protected24H) + } + if snap.ProtectedLookahead != 30_000 { // 仅 A + t.Errorf("ProtectedLookahead=%d, want 30000", snap.ProtectedLookahead) + } + if snap.Protected7D != 30_000 { // 仅 A（200h > 168h） + t.Errorf("Protected7D=%d, want 30000", snap.Protected7D) + } + // lookahead 窗口内到期：B(48h) 和 E(恰好 72h)，C 已到期、D 无锁定不算 + if snap.ExpiringCount != 2 || snap.ExpiringEnergy != 60_000 { + t.Errorf("Expiring=%d/%d, want 2/60000", snap.ExpiringCount, snap.ExpiringEnergy) + } + if snap.NearestExpiryMs != now.UnixMilli()+48*hourMs { + t.Errorf("NearestExpiryMs=%d, want %d", snap.NearestExpiryMs, now.UnixMilli()+48*hourMs) + } + if snap.EnergyLimit != 500_000 || snap.EnergyUsed != 120_000 { + t.Errorf("limit/used=%d/%d, want 500000/120000", snap.EnergyLimit, snap.EnergyUsed) + } +} + +func TestComputeEnergySnapshotZeroWeight(t *testing.T) { + _, err := ComputeEnergySnapshot("T1", nil, + ResourceParams{TotalEnergyLimit: 1, TotalEnergyWeight: 0}, + time.UnixMilli(0), 72*time.Hour) + if err == nil { + t.Fatal("expected error on TotalEnergyWeight=0, got nil") + } +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./chains/tron/ -run TestComputeEnergySnapshot -v` +Expected: FAIL（编译错误：类型未定义） + +- [ ] **Step 3: 实现** + +创建 `chains/tron/energy_calc.go`： + +```go +package tron + +import ( + "fmt" + "time" +) + +// DelegationDetail is one inbound energy delegation as read from chain. +type DelegationDetail struct { + From string + FrozenBalanceSun int64 + ExpireTimeMs int64 // lock expiry, unix ms; 0 = no lock (revocable anytime) +} + +// ResourceParams holds the account/network resource numbers used for the +// sun→energy conversion, from GetAccountResource. +type ResourceParams struct { + EnergyLimit int64 + EnergyUsed int64 + TotalEnergyLimit int64 + TotalEnergyWeight int64 +} + +// EnergySnapshot is the result of one scan of one address. +type EnergySnapshot struct { + Address string `json:"address"` + QueriedAtMs int64 `json:"queriedAtMs"` + EnergyLimit int64 `json:"energyLimit"` + EnergyUsed int64 `json:"energyUsed"` + TotalInboundEnergy int64 `json:"totalInboundEnergy"` + Protected24H int64 `json:"protected24h"` + ProtectedLookahead int64 `json:"protectedLookahead"` + Protected7D int64 `json:"protected7d"` + ExpiringCount int `json:"expiringCount"` + ExpiringEnergy int64 `json:"expiringEnergy"` + NearestExpiryMs int64 `json:"nearestExpiryMs"` +} + +// ComputeEnergySnapshot converts inbound delegations into protected-energy +// metrics. Protected = lock expiry strictly later than the window cutoff; +// expired, expiring-in-window and no-lock delegations are excluded. +func ComputeEnergySnapshot(addr string, dels []DelegationDetail, res ResourceParams, now time.Time, lookahead time.Duration) (*EnergySnapshot, error) { + if res.TotalEnergyWeight <= 0 { + return nil, fmt.Errorf("invalid TotalEnergyWeight %d", res.TotalEnergyWeight) + } + energyPerSun := float64(res.TotalEnergyLimit) / float64(res.TotalEnergyWeight) / 1_000_000 + + nowMs := now.UnixMilli() + cut24 := now.Add(24 * time.Hour).UnixMilli() + cutLook := now.Add(lookahead).UnixMilli() + cut7d := now.Add(7 * 24 * time.Hour).UnixMilli() + + snap := &EnergySnapshot{ + Address: addr, + QueriedAtMs: nowMs, + EnergyLimit: res.EnergyLimit, + EnergyUsed: res.EnergyUsed, + } + for _, d := range dels { + est := int64(float64(d.FrozenBalanceSun) * energyPerSun) + snap.TotalInboundEnergy += est + if d.ExpireTimeMs > cut24 { + snap.Protected24H += est + } + if d.ExpireTimeMs > cutLook { + snap.ProtectedLookahead += est + } + if d.ExpireTimeMs > cut7d { + snap.Protected7D += est + } + if d.ExpireTimeMs > nowMs && d.ExpireTimeMs <= cutLook { + snap.ExpiringCount++ + snap.ExpiringEnergy += est + } + if d.ExpireTimeMs > nowMs && (snap.NearestExpiryMs == 0 || d.ExpireTimeMs < snap.NearestExpiryMs) { + snap.NearestExpiryMs = d.ExpireTimeMs + } + } + return snap, nil +} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `go test ./chains/tron/ -run TestComputeEnergySnapshot -v` +Expected: PASS（两个测试都过） + +- [ ] **Step 5: Commit** + +```bash +git add chains/tron/energy_calc.go chains/tron/energy_calc_test.go +git commit -m "feat(tron): add protected-energy snapshot computation" +``` + +--- + +### Task 3: 状态机 energy_state.go（纯逻辑） + +**Files:** +- Create: `chains/tron/energy_state.go` +- Test: `chains/tron/energy_state_test.go` + +- [ ] **Step 1: 写失败的测试** + +创建 `chains/tron/energy_state_test.go`： + +```go +package tron + +import ( + "testing" + "time" +) + +func pol() ExpiryPolicy { + return ExpiryPolicy{ + ProtectedThreshold: 10_000_000, + RecoveryThreshold: 10_500_000, + RepeatInterval: 12 * time.Hour, + FailureThreshold: 3, + } +} + +func snapWith(protected int64) *EnergySnapshot { + return &EnergySnapshot{ProtectedLookahead: protected} +} + +func TestNextAlertState(t *testing.T) { + base := time.UnixMilli(1_700_000_000_000) + + t.Run("first alert on drop below threshold", func(t *testing.T) { + st, act := NextAlertState(AlertState{}, snapWith(8_000_000), false, pol(), base) + if act != ActionFirstAlert { + t.Fatalf("action=%v, want ActionFirstAlert", act) + } + if st.Status != StatusAlert || st.FirstAlertAtMs != base.UnixMilli() || + st.LastSentAtMs != base.UnixMilli() || st.LastSentProtected != 8_000_000 { + t.Fatalf("state=%+v", st) + } + }) + + t.Run("no repeat within interval", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + _, act := NextAlertState(prev, snapWith(8_100_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone", act) + } + }) + + t.Run("repeat after interval", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(8_000_000), false, pol(), base.Add(12*time.Hour)) + if act != ActionRepeatAlert { + t.Fatalf("action=%v, want ActionRepeatAlert", act) + } + if st.LastSentAtMs != base.Add(12*time.Hour).UnixMilli() { + t.Fatalf("LastSentAtMs not updated: %+v", st) + } + }) + + t.Run("escalate when gap grows over 20 percent", func(t *testing.T) { + // prev gap = 10M-8M = 2M；current 7.5M → gap 2.5M = +25% + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(7_500_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionEscalateAlert { + t.Fatalf("action=%v, want ActionEscalateAlert", act) + } + if st.LastSentProtected != 7_500_000 { + t.Fatalf("LastSentProtected not updated: %+v", st) + } + }) + + t.Run("gap growth under 20 percent does not escalate", func(t *testing.T) { + // prev gap 2M；current 7.7M → gap 2.3M = +15% + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + _, act := NextAlertState(prev, snapWith(7_700_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone", act) + } + }) + + t.Run("recovery buffer holds alert", func(t *testing.T) { + // 10.2M：≥阈值但 <恢复阈值 → 维持 ALERT 不发 + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(10_200_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone || st.Status != StatusAlert { + t.Fatalf("action=%v status=%v, want ActionNone/ALERT", act, st.Status) + } + }) + + t.Run("recovery above recovery threshold", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(10_600_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionRecovery || st.Status != StatusOK { + t.Fatalf("action=%v status=%v, want ActionRecovery/OK", act, st.Status) + } + if st.LastRecoveredAtMs != base.Add(1*time.Hour).UnixMilli() { + t.Fatalf("LastRecoveredAtMs not set: %+v", st) + } + }) + + t.Run("ok stays ok", func(t *testing.T) { + st, act := NextAlertState(AlertState{}, snapWith(20_000_000), false, pol(), base) + if act != ActionNone || st.Status != StatusOK { + t.Fatalf("action=%v status=%v, want ActionNone/OK", act, st.Status) + } + }) + + t.Run("scan failure never triggers energy alert", func(t *testing.T) { + st, act := NextAlertState(AlertState{Status: StatusOK}, nil, true, pol(), base) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone on 1st failure", act) + } + if st.ConsecutiveFails != 1 || st.Status != StatusOK { + t.Fatalf("state=%+v, want fails=1 and baseline kept", st) + } + }) + + t.Run("failure alert on 3rd consecutive failure, only once", func(t *testing.T) { + st := AlertState{Status: StatusAlert, ConsecutiveFails: 2} + st, act := NextAlertState(st, nil, true, pol(), base) + if act != ActionFailureAlert || !st.FailureAlerted { + t.Fatalf("action=%v state=%+v, want ActionFailureAlert", act, st) + } + if st.Status != StatusAlert { + t.Fatalf("baseline lost: %+v", st) + } + _, act = NextAlertState(st, nil, true, pol(), base.Add(time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone on 4th failure (already alerted)", act) + } + }) + + t.Run("success after unknown resets fails and rejudges against baseline", func(t *testing.T) { + // 之前是 ALERT，UNKNOWN 若干轮后恢复查询且低于阈值 → 不是首报（基线还是 ALERT）， + // 距上次发送超 12h → 重复提醒 + st := AlertState{Status: StatusAlert, ConsecutiveFails: 5, FailureAlerted: true, + LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(st, snapWith(8_000_000), false, pol(), base.Add(13*time.Hour)) + if act != ActionRepeatAlert { + t.Fatalf("action=%v, want ActionRepeatAlert", act) + } + if st.ConsecutiveFails != 0 || st.FailureAlerted { + t.Fatalf("failure tracking not reset: %+v", st) + } + }) +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./chains/tron/ -run TestNextAlertState -v` +Expected: FAIL（编译错误：类型未定义） + +- [ ] **Step 3: 实现** + +创建 `chains/tron/energy_state.go`： + +```go +package tron + +import "time" + +type AlertStatus string + +const ( + StatusOK AlertStatus = "OK" + StatusAlert AlertStatus = "ALERT" + StatusUnknown AlertStatus = "UNKNOWN" +) + +type AlertAction int + +const ( + ActionNone AlertAction = iota + ActionFirstAlert + ActionRepeatAlert + ActionEscalateAlert + ActionRecovery + ActionFailureAlert +) + +// ExpiryPolicy is the per-address alerting policy (already defaulted). +type ExpiryPolicy struct { + ProtectedThreshold int64 + RecoveryThreshold int64 + RepeatInterval time.Duration + FailureThreshold int +} + +// AlertState is the persisted alert-dedup state for one address. +// Status only holds the last confirmed OK/ALERT baseline; scan failures are +// tracked via ConsecutiveFails and never overwrite the baseline, so an API +// outage can neither trigger nor clear an energy alert. +type AlertState struct { + Status AlertStatus `json:"status,omitempty"` + FirstAlertAtMs int64 `json:"firstAlertAtMs,omitempty"` + LastSentAtMs int64 `json:"lastSentAtMs,omitempty"` + LastRecoveredAtMs int64 `json:"lastRecoveredAtMs,omitempty"` + LastSentProtected int64 `json:"lastSentProtected,omitempty"` + ConsecutiveFails int `json:"consecutiveFails,omitempty"` + FailureAlerted bool `json:"failureAlerted,omitempty"` +} + +// NextAlertState advances the state machine for one scan result and returns +// the action to perform. snap may be nil when scanFailed is true. +func NextAlertState(st AlertState, snap *EnergySnapshot, scanFailed bool, pol ExpiryPolicy, now time.Time) (AlertState, AlertAction) { + if scanFailed { + st.ConsecutiveFails++ + if st.ConsecutiveFails >= pol.FailureThreshold && !st.FailureAlerted { + st.FailureAlerted = true + return st, ActionFailureAlert + } + return st, ActionNone + } + + st.ConsecutiveFails = 0 + st.FailureAlerted = false + nowMs := now.UnixMilli() + + if snap.ProtectedLookahead < pol.ProtectedThreshold { + if st.Status != StatusAlert { // first alert (baseline was OK/empty) + st.Status = StatusAlert + st.FirstAlertAtMs = nowMs + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionFirstAlert + } + prevGap := pol.ProtectedThreshold - st.LastSentProtected + curGap := pol.ProtectedThreshold - snap.ProtectedLookahead + if prevGap > 0 && float64(curGap) > float64(prevGap)*1.2 { + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionEscalateAlert + } + if nowMs-st.LastSentAtMs >= pol.RepeatInterval.Milliseconds() { + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionRepeatAlert + } + return st, ActionNone + } + + if st.Status == StatusAlert { + if snap.ProtectedLookahead >= pol.RecoveryThreshold { + st.Status = StatusOK + st.LastRecoveredAtMs = nowMs + return st, ActionRecovery + } + return st, ActionNone // recovery buffer: stay ALERT silently + } + st.Status = StatusOK + return st, ActionNone +} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `go test ./chains/tron/ -run TestNextAlertState -v` +Expected: PASS（全部子测试） + +- [ ] **Step 5: Commit** + +```bash +git add chains/tron/energy_state.go chains/tron/energy_state_test.go +git commit -m "feat(tron): add energy expiry alert state machine" +``` + +--- + +### Task 4: 持久化 energy_store.go + +**Files:** +- Create: `chains/tron/energy_store.go` +- Test: `chains/tron/energy_store_test.go` + +- [ ] **Step 1: 写失败的测试** + +创建 `chains/tron/energy_store_test.go`： + +```go +package tron + +import ( + "os" + "path/filepath" + "testing" +) + +func TestEnergyStateRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "sub", "energy_state_T1.json") + + st, err := LoadEnergyState(path, "T1") + if err != nil { + t.Fatalf("load missing file: %v", err) + } + if st.Address != "T1" || len(st.Snapshots) != 0 { + t.Fatalf("fresh state=%+v", st) + } + + st.Alert = AlertState{Status: StatusAlert, LastSentProtected: 123} + st.Append(SnapshotEntry{AtMs: 1, Status: StatusAlert, Snap: &EnergySnapshot{Address: "T1"}}) + if err := SaveEnergyState(path, st); err != nil { + t.Fatalf("save: %v", err) + } + + got, err := LoadEnergyState(path, "T1") + if err != nil { + t.Fatalf("reload: %v", err) + } + if got.Alert.Status != StatusAlert || got.Alert.LastSentProtected != 123 || len(got.Snapshots) != 1 { + t.Fatalf("reloaded=%+v", got) + } +} + +func TestEnergyStateHistoryCap(t *testing.T) { + st := &StoredEnergyState{Address: "T1"} + for i := 0; i < maxSnapshotHistory+10; i++ { + st.Append(SnapshotEntry{AtMs: int64(i)}) + } + if len(st.Snapshots) != maxSnapshotHistory { + t.Fatalf("len=%d, want %d", len(st.Snapshots), maxSnapshotHistory) + } + if st.Snapshots[0].AtMs != 10 { + t.Fatalf("oldest=%d, want 10 (ring dropped head)", st.Snapshots[0].AtMs) + } +} + +func TestLoadEnergyStateCorrupt(t *testing.T) { + path := filepath.Join(t.TempDir(), "energy_state_T1.json") + if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil { + t.Fatal(err) + } + st, err := LoadEnergyState(path, "T1") + if err == nil { + t.Fatal("expected error for corrupt file") + } + if st == nil || st.Address != "T1" { + t.Fatalf("must still return usable fresh state, got %+v", st) + } +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./chains/tron/ -run TestEnergyState -v` +Expected: FAIL（编译错误） + +- [ ] **Step 3: 实现** + +创建 `chains/tron/energy_store.go`： + +```go +package tron + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +// maxSnapshotHistory bounds the persisted scan history: 168 hourly scans ≈ 7 days. +const maxSnapshotHistory = 168 + +// SnapshotEntry is one scan result kept for audit. +type SnapshotEntry struct { + AtMs int64 `json:"atMs"` + Status AlertStatus `json:"status"` + Err string `json:"err,omitempty"` + Snap *EnergySnapshot `json:"snapshot,omitempty"` +} + +// StoredEnergyState is the on-disk state for one monitored address. +type StoredEnergyState struct { + Address string `json:"address"` + Alert AlertState `json:"alert"` + Snapshots []SnapshotEntry `json:"snapshots"` +} + +// Append adds an entry, dropping the oldest beyond maxSnapshotHistory. +func (s *StoredEnergyState) Append(e SnapshotEntry) { + s.Snapshots = append(s.Snapshots, e) + if n := len(s.Snapshots) - maxSnapshotHistory; n > 0 { + s.Snapshots = append(s.Snapshots[:0:0], s.Snapshots[n:]...) + } +} + +// LoadEnergyState reads state from path. A missing file yields a fresh state +// and nil error; a corrupt file yields a fresh state AND the error, so the +// caller can log/alarm but keep monitoring. +func LoadEnergyState(path, address string) (*StoredEnergyState, error) { + fresh := &StoredEnergyState{Address: address} + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return fresh, nil + } + return fresh, err + } + st := &StoredEnergyState{} + if err := json.Unmarshal(data, st); err != nil { + return fresh, fmt.Errorf("corrupt energy state %s: %w", path, err) + } + st.Address = address + return st, nil +} + +// SaveEnergyState writes state atomically (tmp file + rename), creating the +// parent directory when needed. +func SaveEnergyState(path string, st *StoredEnergyState) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + data, err := json.MarshalIndent(st, "", " ") + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `go test ./chains/tron/ -run TestEnergyState -v && go test ./chains/tron/ -run TestLoadEnergyStateCorrupt -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add chains/tron/energy_store.go chains/tron/energy_store_test.go +git commit -m "feat(tron): add JSON persistence for energy expiry state" +``` + +--- + +### Task 5: 查询层 — con.go 入站代理封装 + +**Files:** +- Modify: `chains/tron/con.go` +- Test: `chains/tron/con_test.go`（新建，仅测 withRetry） + +**背景（务必理解再动手）：** SDK 现成的 `GetDelegatedResourcesV2` 遍历的是 `ToAccounts`（出站方向，查"该地址代理给了谁"），与需求相反。必须绕过它直接用 raw stub `c.cli.Client`（类型 `api.WalletClient`）：先 `GetDelegatedResourceAccountIndexV2` 取 `FromAccounts`，再对每个 from 调 `GetDelegatedResourceV2(from→target)`。 + +- [ ] **Step 1: 写失败的测试（重试助手）** + +创建 `chains/tron/con_test.go`： + +```go +package tron + +import ( + "errors" + "testing" + "time" +) + +func TestWithRetry(t *testing.T) { + t.Run("succeeds after transient failures", func(t *testing.T) { + calls := 0 + err := withRetry(3, time.Millisecond, func() error { + calls++ + if calls < 3 { + return errors.New("transient") + } + return nil + }) + if err != nil || calls != 3 { + t.Fatalf("err=%v calls=%d, want nil/3", err, calls) + } + }) + + t.Run("returns last error when exhausted", func(t *testing.T) { + calls := 0 + err := withRetry(3, time.Millisecond, func() error { + calls++ + return errors.New("boom") + }) + if err == nil || calls != 3 { + t.Fatalf("err=%v calls=%d, want error/3", err, calls) + } + }) +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./chains/tron/ -run TestWithRetry -v` +Expected: FAIL（withRetry 未定义） + +- [ ] **Step 3: 实现** + +在 `chains/tron/con.go` 追加（import 需增加 `context`、`github.com/lbtsm/gotron-sdk/pkg/client`、`github.com/lbtsm/gotron-sdk/pkg/common`、`github.com/lbtsm/gotron-sdk/pkg/proto/api`、`github.com/pkg/errors`；注意 `errors` 用 pkg/errors，与项目其他文件一致）： + +```go +const ( + delegationRequestGap = 500 * time.Millisecond + queryMaxRetries = 3 + queryRetryBackoff = time.Second + queryTimeout = 15 * time.Second +) + +// withRetry runs do up to attempts times with exponential backoff. +func withRetry(attempts int, backoff time.Duration, do func() error) error { + var err error + for i := 0; i < attempts; i++ { + if err = do(); err == nil { + return nil + } + if i < attempts-1 { + time.Sleep(backoff) + backoff *= 2 + } + } + return err +} + +// InboundDelegations lists every energy delegation TO target (Stake 2.0). +// Any final failure returns an error — callers must treat the whole scan as +// UNKNOWN rather than compute from partial data. +func (c *Connection) InboundDelegations(target string) ([]DelegationDetail, error) { + targetBytes, err := common.DecodeCheck(target) + if err != nil { + return nil, errors.Wrapf(err, "decode address %s", target) + } + + index, err := c.delegationIndex(targetBytes) + if err != nil { + return nil, errors.Wrap(err, "GetDelegatedResourceAccountIndexV2") + } + + details := make([]DelegationDetail, 0, len(index.GetFromAccounts())) + for i, from := range index.GetFromAccounts() { + if i > 0 { + time.Sleep(delegationRequestGap) + } + list, err := c.delegationDetail(from, targetBytes) + if err != nil { + return nil, errors.Wrapf(err, "GetDelegatedResourceV2 from %s", common.EncodeCheck(from)) + } + for _, d := range list.GetDelegatedResource() { + if d.GetFrozenBalanceForEnergy() <= 0 { + continue // bandwidth-only delegation + } + details = append(details, DelegationDetail{ + From: common.EncodeCheck(d.GetFrom()), + FrozenBalanceSun: d.GetFrozenBalanceForEnergy(), + ExpireTimeMs: d.GetExpireTimeForEnergy(), + }) + } + } + return details, nil +} + +func (c *Connection) delegationIndex(target []byte) (*core.DelegatedResourceAccountIndex, error) { + var index *core.DelegatedResourceAccountIndex + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + ctx, cancel := context.WithTimeout(context.Background(), queryTimeout) + defer cancel() + var e error + index, e = c.cli.Client.GetDelegatedResourceAccountIndexV2(ctx, client.GetMessageBytes(target)) + return e + }) + return index, err +} + +func (c *Connection) delegationDetail(from, to []byte) (*api.DelegatedResourceList, error) { + var list *api.DelegatedResourceList + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + ctx, cancel := context.WithTimeout(context.Background(), queryTimeout) + defer cancel() + var e error + list, e = c.cli.Client.GetDelegatedResourceV2(ctx, &api.DelegatedResourceMessage{ + FromAddress: from, + ToAddress: to, + }) + return e + }) + return list, err +} + +// EnergyResourceParams reads the numbers needed for sun→energy conversion. +func (c *Connection) EnergyResourceParams(target string) (ResourceParams, error) { + var params ResourceParams + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + res, e := c.cli.GetAccountResource(target) + if e != nil { + return e + } + params = ResourceParams{ + EnergyLimit: res.GetEnergyLimit(), + EnergyUsed: res.GetEnergyUsed(), + TotalEnergyLimit: res.GetTotalEnergyLimit(), + TotalEnergyWeight: res.GetTotalEnergyWeight(), + } + return nil + }) + return params, err +} +``` + +import 块还需 `github.com/lbtsm/gotron-sdk/pkg/proto/core`。注意 `core.DelegatedResourceAccountIndex` 是 stub 返回类型（在 `Tron.pb.go`，有 `GetFromAccounts() [][]byte` 方法）——如编译报类型不匹配，以 `api.WalletClient` 接口中 `GetDelegatedResourceAccountIndexV2` 的实际返回类型为准修正。 + +- [ ] **Step 4: 运行测试与编译** + +Run: `go test ./chains/tron/ -run TestWithRetry -v && go build ./chains/tron/` +Expected: 测试 PASS，编译通过 + +- [ ] **Step 5: Commit** + +```bash +git add chains/tron/con.go chains/tron/con_test.go +git commit -m "feat(tron): add inbound delegation gRPC queries with retry" +``` + +--- + +### Task 6: 检查器 energy.go（调度、文案、报警） + +**Files:** +- Create: `chains/tron/energy.go` +- Test: `chains/tron/energy_test.go` + +- [ ] **Step 1: 写失败的测试** + +创建 `chains/tron/energy_test.go`： + +```go +package tron + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/ChainSafe/log15" + "github.com/mapprotocol/monitor/internal/config" +) + +type fakeQuerier struct { + dels []DelegationDetail + res ResourceParams + err error +} + +func (f *fakeQuerier) InboundDelegations(string) ([]DelegationDetail, error) { + if f.err != nil { + return nil, f.err + } + return f.dels, nil +} + +func (f *fakeQuerier) EnergyResourceParams(string) (ResourceParams, error) { + if f.err != nil { + return ResourceParams{}, f.err + } + return f.res, nil +} + +func testEnergyCfg() config.Energy { + en := config.Energy{Address: "TTest1", ProtectedThreshold: 100_000} + en.ApplyExpiryDefaults() + return en +} + +func newTestChecker(t *testing.T, q energyQuerier) (*expiryChecker, *[]string) { + t.Helper() + var sent []string + c := newExpiryChecker(log15.New(), q, "tron", t.TempDir(), + func(_ context.Context, msg string) { sent = append(sent, msg) }) + return c, &sent +} + +func TestRunOnceFirstAlertAndPersist(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + q := &fakeQuerier{ + // 30 TRX 无锁定 → protected 0 < 100_000 阈值 + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 30_000_000, ExpireTimeMs: 0}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c, sent := newTestChecker(t, q) + en := testEnergyCfg() + + c.runOnce(en, now) + + if len(*sent) != 1 || !strings.Contains((*sent)[0], "TRON Energy 预警") { + t.Fatalf("sent=%v, want one first alert", *sent) + } + // 状态已持久化，重启（新 checker）后同样低于阈值不再首报 + c2, sent2 := newTestChecker(t, q) + c2.dir = c.dir + c2.runOnce(en, now.Add(time.Hour)) + if len(*sent2) != 0 { + t.Fatalf("sent after restart=%v, want dedup (no resend within 12h)", *sent2) + } + st, err := LoadEnergyState(filepath.Join(c.dir, "energy_state_TTest1.json"), "TTest1") + if err != nil || st.Alert.Status != StatusAlert || len(st.Snapshots) != 1 { + t.Fatalf("persisted state=%+v err=%v", st, err) + } +} + +func TestRunOnceFailureIsUnknownNotZero(t *testing.T) { + q := &fakeQuerier{err: errors.New("rpc down")} + c, sent := newTestChecker(t, q) + en := testEnergyCfg() + now := time.UnixMilli(1_700_000_000_000) + + c.runOnce(en, now) + c.runOnce(en, now.Add(time.Hour)) + if len(*sent) != 0 { + t.Fatalf("sent=%v, want none before 3rd failure", *sent) + } + c.runOnce(en, now.Add(2*time.Hour)) + if len(*sent) != 1 || !strings.Contains((*sent)[0], "监控异常") { + t.Fatalf("sent=%v, want one failure alert", *sent) + } + if strings.Contains((*sent)[0], "预警]") { + t.Fatalf("failure alert must not look like an energy alert: %v", *sent) + } +} + +func TestRunOnceRecovery(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + low := &fakeQuerier{ + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 30_000_000, ExpireTimeMs: 0}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c, sent := newTestChecker(t, low) + en := testEnergyCfg() + c.runOnce(en, now) // first alert + + // 换成充足数据：10000 TRX 锁到 30 天后 → 300_000 > recovery 105_000 + c.q = &fakeQuerier{ + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 10_000_000_000, + ExpireTimeMs: now.Add(30 * 24 * time.Hour).UnixMilli()}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c.runOnce(en, now.Add(time.Hour)) + if len(*sent) != 2 || !strings.Contains((*sent)[1], "恢复") { + t.Fatalf("sent=%v, want alert then recovery", *sent) + } +} + +func TestFmtEnergyInt(t *testing.T) { + tests := []struct { + in int64 + want string + }{{0, "0"}, {999, "999"}, {1000, "1,000"}, {10500000, "10,500,000"}, {-1234, "-1,234"}} + for _, tt := range tests { + if got := fmtEnergyInt(tt.in); got != tt.want { + t.Errorf("fmtEnergyInt(%d)=%q, want %q", tt.in, got, tt.want) + } + } +} +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `go test ./chains/tron/ -run 'TestRunOnce|TestFmtEnergyInt' -v` +Expected: FAIL（编译错误） + +- [ ] **Step 3: 实现** + +创建 `chains/tron/energy.go`： + +```go +package tron + +import ( + "context" + "fmt" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/ChainSafe/log15" + "github.com/mapprotocol/monitor/internal/config" +) + +// energyQuerier is the chain-access surface the checker needs; *Connection +// implements it, tests use a fake. +type energyQuerier interface { + InboundDelegations(target string) ([]DelegationDetail, error) + EnergyResourceParams(target string) (ResourceParams, error) +} + +const energyFailureThreshold = 3 + +// displayLoc is the timezone used in alert messages (storage stays UTC ms). +var displayLoc = func() *time.Location { + loc, err := time.LoadLocation("Asia/Singapore") + if err != nil { + return time.UTC + } + return loc +}() + +// expiryChecker runs protected-energy scans and drives alerting for the +// addresses configured with a protectedThreshold. +type expiryChecker struct { + log log15.Logger + q energyQuerier + chain string + dir string // state-file directory + states map[string]*StoredEnergyState + alarm func(ctx context.Context, msg string) +} + +func newExpiryChecker(log log15.Logger, q energyQuerier, chainName, dir string, + alarm func(ctx context.Context, msg string)) *expiryChecker { + return &expiryChecker{ + log: log, + q: q, + chain: chainName, + dir: dir, + states: make(map[string]*StoredEnergyState), + alarm: alarm, + } +} + +func (c *expiryChecker) statePath(addr string) string { + return filepath.Join(c.dir, "energy_state_"+addr+".json") +} + +func (c *expiryChecker) state(addr string) *StoredEnergyState { + if st, ok := c.states[addr]; ok { + return st + } + st, err := LoadEnergyState(c.statePath(addr), addr) + if err != nil { + c.log.Error("EnergyExpiry state file unreadable, starting fresh", "addr", addr, "err", err) + } + c.states[addr] = st + return st +} + +// runOnce performs one scan+judge+persist+alert cycle for one address. +// en must already have ApplyExpiryDefaults applied. +func (c *expiryChecker) runOnce(en config.Energy, now time.Time) { + st := c.state(en.Address) + pol := ExpiryPolicy{ + ProtectedThreshold: en.ProtectedThreshold, + RecoveryThreshold: en.RecoveryThreshold, + RepeatInterval: time.Duration(en.RepeatIntervalHours) * time.Hour, + FailureThreshold: energyFailureThreshold, + } + lookahead := time.Duration(en.LookaheadHours) * time.Hour + + snap, scanErr := c.scan(en.Address, now, lookahead) + newAlert, action := NextAlertState(st.Alert, snap, scanErr != nil, pol, now) + + entry := SnapshotEntry{AtMs: now.UnixMilli(), Snap: snap} + if scanErr != nil { + entry.Status = StatusUnknown + entry.Err = scanErr.Error() + c.log.Error("EnergyExpiry scan failed", "addr", en.Address, + "consecutiveFails", newAlert.ConsecutiveFails, "err", scanErr) + } else { + entry.Status = newAlert.Status + c.log.Info("EnergyExpiry scan", "addr", en.Address, + "protected", snap.ProtectedLookahead, "threshold", en.ProtectedThreshold, + "expiring", snap.ExpiringCount, "status", newAlert.Status) + } + st.Alert = newAlert + st.Append(entry) + if err := SaveEnergyState(c.statePath(en.Address), st); err != nil { + c.log.Error("EnergyExpiry save state failed", "addr", en.Address, "err", err) + } + + switch action { + case ActionFirstAlert, ActionRepeatAlert, ActionEscalateAlert: + c.alarm(context.Background(), formatEnergyAlert(c.chain, en, snap, now, lookahead)) + case ActionRecovery: + c.alarm(context.Background(), formatEnergyRecovery(c.chain, en, snap)) + case ActionFailureAlert: + c.alarm(context.Background(), formatEnergyFailure(c.chain, en.Address, newAlert.ConsecutiveFails, scanErr)) + } +} + +func (c *expiryChecker) scan(addr string, now time.Time, lookahead time.Duration) (*EnergySnapshot, error) { + dels, err := c.q.InboundDelegations(addr) + if err != nil { + return nil, err + } + res, err := c.q.EnergyResourceParams(addr) + if err != nil { + return nil, err + } + return ComputeEnergySnapshot(addr, dels, res, now, lookahead) +} + +func fmtEnergyTime(ms int64) string { + if ms <= 0 { + return "-" + } + return time.UnixMilli(ms).In(displayLoc).Format("2006-01-02 15:04 MST") +} + +func fmtEnergyInt(n int64) string { + neg := n < 0 + if neg { + n = -n + } + s := strconv.FormatInt(n, 10) + var b strings.Builder + pre := len(s) % 3 + if pre > 0 { + b.WriteString(s[:pre]) + } + for i := pre; i < len(s); i += 3 { + if b.Len() > 0 { + b.WriteByte(',') + } + b.WriteString(s[i : i+3]) + } + if neg { + return "-" + b.String() + } + return b.String() +} + +func formatEnergyAlert(chainName string, en config.Energy, snap *EnergySnapshot, now time.Time, lookahead time.Duration) string { + return fmt.Sprintf(`[TRON Energy 预警] +链：%s +地址：%s +预测时间：%s +%d 小时后受保护 Energy：%s +配置阈值：%s +缺口：%s +未来 %d 小时到期：%d 笔，共约 %s Energy +最近到期：%s +查询时间：%s`, + chainName, en.Address, + fmtEnergyTime(now.Add(lookahead).UnixMilli()), + en.LookaheadHours, fmtEnergyInt(snap.ProtectedLookahead), + fmtEnergyInt(en.ProtectedThreshold), + fmtEnergyInt(en.ProtectedThreshold-snap.ProtectedLookahead), + en.LookaheadHours, snap.ExpiringCount, fmtEnergyInt(snap.ExpiringEnergy), + fmtEnergyTime(snap.NearestExpiryMs), + fmtEnergyTime(snap.QueriedAtMs)) +} + +func formatEnergyRecovery(chainName string, en config.Energy, snap *EnergySnapshot) string { + return fmt.Sprintf(`[TRON Energy 恢复] +链：%s +地址：%s +%d 小时后受保护 Energy：%s（≥ 恢复阈值 %s） +查询时间：%s`, + chainName, en.Address, + en.LookaheadHours, fmtEnergyInt(snap.ProtectedLookahead), + fmtEnergyInt(en.RecoveryThreshold), + fmtEnergyTime(snap.QueriedAtMs)) +} + +func formatEnergyFailure(chainName, addr string, fails int, scanErr error) string { + return fmt.Sprintf(`[TRON Energy 监控异常] +链：%s +地址：%s +连续 %d 次查询失败，受保护 Energy 状态未知（不代表 Energy 为 0） +最近错误：%v`, + chainName, addr, fails, scanErr) +} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `go test ./chains/tron/ -run 'TestRunOnce|TestFmtEnergyInt' -v` +Expected: PASS（4 个测试全过） + +- [ ] **Step 5: Commit** + +```bash +git add chains/tron/energy.go chains/tron/energy_test.go +git commit -m "feat(tron): add energy expiry checker with dedup alerting" +``` + +--- + +### Task 7: 接入 Monitor.Sync 并全量验证 + +**Files:** +- Modify: `chains/tron/monitor.go`（Sync 方法，约第 40 行） +- Modify: `chains/tron/energy.go`（追加 energyExpirySync 方法） + +- [ ] **Step 1: 在 energy.go 追加调度循环** + +在 `chains/tron/energy.go` 末尾追加（import 增加 `github.com/mapprotocol/monitor/internal/chain`、`github.com/mapprotocol/monitor/pkg/util`、`github.com/pkg/errors`；注意 fmt 的 errors 冲突——本文件用 pkg/errors 只在此函数）： + +```go +// energyExpiryTick is how often the scheduler wakes to check per-address due +// times; actual scan cadence is each address's CheckIntervalMinutes. +const energyExpiryTick = time.Minute + +// energyExpirySync schedules expiry scans for all enabled energy entries. +// It re-reads the config snapshot every tick, so hot-reloaded thresholds and +// newly added addresses are picked up without restart. +func (m *Monitor) energyExpirySync() error { + checker := newExpiryChecker(m.Log, m.conn, m.Cfg.Name, m.Cfg.KeystorePath, util.Alarm) + nextRun := make(map[string]time.Time) + for { + select { + case <-m.Stop: + return errors.New("energy expiry polling terminated") + default: + snap := m.Snapshot() + for _, en := range snap.Energies { + if en.ProtectedThreshold <= 0 { + continue + } + en.ApplyExpiryDefaults() + now := time.Now() + if now.Before(nextRun[en.Address]) { + continue + } + checker.runOnce(en, now) + nextRun[en.Address] = now.Add(time.Duration(en.CheckIntervalMinutes) * time.Minute) + } + if !chain.SleepWithStop(m.Stop, energyExpiryTick) { + return errors.New("energy expiry polling terminated") + } + } + } +} +``` + +注意：`energy.go` 顶部 import 已有标准库 `fmt`；`errors` 引入的是 `github.com/pkg/errors`（项目惯例）。 + +- [ ] **Step 2: 修改 monitor.go 的 Sync 启动检查器** + +`chains/tron/monitor.go` 的 `Sync()`（原第 40-51 行）改为： + +```go +func (m *Monitor) Sync() error { + m.Log.Debug("Starting listener...") + m.Wg.Add(1) + go func() { + defer m.Wg.Done() + if err := m.sync(); err != nil { + m.Log.Error("Polling Account balance failed", "err", err) + } + }() + + m.Wg.Add(1) + go func() { + defer m.Wg.Done() + if err := m.energyExpirySync(); err != nil { + m.Log.Error("Energy expiry polling stopped", "err", err) + } + }() + + return nil +} +``` + +- [ ] **Step 3: 编译与全量测试** + +Run: `go build ./... && go test ./chains/tron/... ./internal/config/...` +Expected: 编译通过；chains/tron 全部 PASS；internal/config 中若有与本改动无关的既有失败（该目录有用户未提交改动），记录但不阻塞。 + +- [ ] **Step 4: Commit** + +```bash +git add chains/tron/energy.go chains/tron/monitor.go +git commit -m "feat(tron): wire energy expiry checker into monitor sync" +``` + +- [ ] **Step 5: 配置样例说明（写入 README）** + +在 `README.md` 的 Options 代码块后追加： + +```markdown +## TRON Energy Expiry Monitoring + +Add `protectedThreshold` to a tron chain's `energy` entry to enable +Stake 2.0 delegation-expiry alerting (protected energy = inbound delegations +whose lock expires strictly after now+lookahead): + +```shell +"energy": [{ + "address": "TT6GDYkpHPVk24w9he9pavbagtzqBRS3XP", + "waterline": 100000, // existing: current remaining-energy alarm + "protectedThreshold": 10000000, // alert when protected energy drops below + "recoveryThreshold": 10500000, // optional, default = protected × 1.05 + "lookaheadHours": 72, // optional, default 72 + "checkIntervalMinutes": 60, // optional, default 60 + "repeatIntervalHours": 12 // optional, default 12 +}] +``` + +State files are written to `/energy_state_
.json`. +Scan failures alarm separately as "监控异常" and never count as zero energy. +``` + +- [ ] **Step 6: Commit** + +```bash +git add README.md +git commit -m "docs: document tron energy expiry monitoring config" +``` + +--- + +## 验收对照（需求第 11 章 → 实现位置) + +| 验收项 | 实现 | +|---|---| +| 1 列出全部入站代理 | Task 5 `InboundDelegations` | +| 2 明细含代理方/质押/估算/到期 | Task 2 `DelegationDetail`/`EnergySnapshot` | +| 3 排除 72h 内到期与无锁定 | Task 2 `ComputeEnergySnapshot`（严格晚于 cutoff） | +| 4 一个周期内报警 | Task 7 调度 + Task 3 `ActionFirstAlert` | +| 5 不重复轰炸 | Task 3 `RepeatInterval` 去重 | +| 6 恢复通知 | Task 3 `ActionRecovery`（恢复阈值缓冲） | +| 7 API 失败不误报 0 | Task 3 scanFailed 分支 + Task 5 整轮报错 + Task 6 UNKNOWN 文案 | +| 8 历史可追溯 | Task 4 快照环形缓冲（168 轮） | +| 9 凭据不落盘 | 无新增凭据；沿用 hooks 环境变量 | From 41783a4c2d37c61388fd5fcb73fa51c0bbbea67f Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:07:43 +0800 Subject: [PATCH 03/13] feat(config): add tron energy expiry monitoring fields to Energy Co-Authored-By: Claude Fable 5 --- internal/config/config.go | 29 ++++++++++++++ internal/config/energy_test.go | 70 ++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+) create mode 100644 internal/config/energy_test.go diff --git a/internal/config/config.go b/internal/config/config.go index 7b3187f..0168d7e 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -71,6 +71,35 @@ type ContractToken struct { type Energy struct { Address string `json:"address"` Waterline int64 `json:"waterline"` + // Expiry monitoring (Stake 2.0 delegation lock). Disabled when + // ProtectedThreshold <= 0; remaining fields are defaulted by + // ApplyExpiryDefaults. + ProtectedThreshold int64 `json:"protectedThreshold"` + RecoveryThreshold int64 `json:"recoveryThreshold"` + LookaheadHours int64 `json:"lookaheadHours"` + CheckIntervalMinutes int64 `json:"checkIntervalMinutes"` + RepeatIntervalHours int64 `json:"repeatIntervalHours"` +} + +// ApplyExpiryDefaults fills unset expiry-monitoring fields. No-op when the +// feature is disabled (ProtectedThreshold <= 0). +func (e *Energy) ApplyExpiryDefaults() { + if e.ProtectedThreshold <= 0 { + return + } + if e.RecoveryThreshold <= 0 { + // protected × 1.05, rounded up + e.RecoveryThreshold = e.ProtectedThreshold + (e.ProtectedThreshold+19)/20 + } + if e.LookaheadHours <= 0 { + e.LookaheadHours = 72 + } + if e.CheckIntervalMinutes <= 0 { + e.CheckIntervalMinutes = 60 + } + if e.RepeatIntervalHours <= 0 { + e.RepeatIntervalHours = 12 + } } type EthToken struct { diff --git a/internal/config/energy_test.go b/internal/config/energy_test.go new file mode 100644 index 0000000..5dd335a --- /dev/null +++ b/internal/config/energy_test.go @@ -0,0 +1,70 @@ +package config + +import "testing" + +func TestEnergyApplyExpiryDefaults(t *testing.T) { + tests := []struct { + name string + in Energy + want Energy + }{ + { + name: "disabled when protectedThreshold is zero", + in: Energy{Address: "T1", Waterline: 100}, + want: Energy{Address: "T1", Waterline: 100}, + }, + { + name: "fills all defaults", + in: Energy{Address: "T1", ProtectedThreshold: 10000000}, + want: Energy{ + Address: "T1", + ProtectedThreshold: 10000000, + RecoveryThreshold: 10500000, // ×1.05 向上取整 + LookaheadHours: 72, + CheckIntervalMinutes: 60, + RepeatIntervalHours: 12, + }, + }, + { + name: "recovery rounds up", + in: Energy{Address: "T1", ProtectedThreshold: 3}, + want: Energy{ + Address: "T1", + ProtectedThreshold: 3, + RecoveryThreshold: 4, // 3*1.05=3.15 → ceil 4 + LookaheadHours: 72, + CheckIntervalMinutes: 60, + RepeatIntervalHours: 12, + }, + }, + { + name: "explicit values are kept", + in: Energy{ + Address: "T1", + ProtectedThreshold: 100, + RecoveryThreshold: 120, + LookaheadHours: 24, + CheckIntervalMinutes: 30, + RepeatIntervalHours: 6, + }, + want: Energy{ + Address: "T1", + ProtectedThreshold: 100, + RecoveryThreshold: 120, + LookaheadHours: 24, + CheckIntervalMinutes: 30, + RepeatIntervalHours: 6, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tt.in + got.ApplyExpiryDefaults() + if got != tt.want { + t.Fatalf("ApplyExpiryDefaults()=%+v, want %+v", got, tt.want) + } + }) + } +} From c00c95fd5069595df5a5f302f6d1329158bdcbcb Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:08:50 +0800 Subject: [PATCH 04/13] feat(tron): add protected-energy snapshot computation Co-Authored-By: Claude Fable 5 --- chains/tron/energy_calc.go | 80 +++++++++++++++++++++++++++++++++ chains/tron/energy_calc_test.go | 65 +++++++++++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 chains/tron/energy_calc.go create mode 100644 chains/tron/energy_calc_test.go diff --git a/chains/tron/energy_calc.go b/chains/tron/energy_calc.go new file mode 100644 index 0000000..e767108 --- /dev/null +++ b/chains/tron/energy_calc.go @@ -0,0 +1,80 @@ +package tron + +import ( + "fmt" + "time" +) + +// DelegationDetail is one inbound energy delegation as read from chain. +type DelegationDetail struct { + From string + FrozenBalanceSun int64 + ExpireTimeMs int64 // lock expiry, unix ms; 0 = no lock (revocable anytime) +} + +// ResourceParams holds the account/network resource numbers used for the +// sun→energy conversion, from GetAccountResource. +type ResourceParams struct { + EnergyLimit int64 + EnergyUsed int64 + TotalEnergyLimit int64 + TotalEnergyWeight int64 +} + +// EnergySnapshot is the result of one scan of one address. +type EnergySnapshot struct { + Address string `json:"address"` + QueriedAtMs int64 `json:"queriedAtMs"` + EnergyLimit int64 `json:"energyLimit"` + EnergyUsed int64 `json:"energyUsed"` + TotalInboundEnergy int64 `json:"totalInboundEnergy"` + Protected24H int64 `json:"protected24h"` + ProtectedLookahead int64 `json:"protectedLookahead"` + Protected7D int64 `json:"protected7d"` + ExpiringCount int `json:"expiringCount"` + ExpiringEnergy int64 `json:"expiringEnergy"` + NearestExpiryMs int64 `json:"nearestExpiryMs"` +} + +// ComputeEnergySnapshot converts inbound delegations into protected-energy +// metrics. Protected = lock expiry strictly later than the window cutoff; +// expired, expiring-in-window and no-lock delegations are excluded. +func ComputeEnergySnapshot(addr string, dels []DelegationDetail, res ResourceParams, now time.Time, lookahead time.Duration) (*EnergySnapshot, error) { + if res.TotalEnergyWeight <= 0 { + return nil, fmt.Errorf("invalid TotalEnergyWeight %d", res.TotalEnergyWeight) + } + energyPerSun := float64(res.TotalEnergyLimit) / float64(res.TotalEnergyWeight) / 1_000_000 + + nowMs := now.UnixMilli() + cut24 := now.Add(24 * time.Hour).UnixMilli() + cutLook := now.Add(lookahead).UnixMilli() + cut7d := now.Add(7 * 24 * time.Hour).UnixMilli() + + snap := &EnergySnapshot{ + Address: addr, + QueriedAtMs: nowMs, + EnergyLimit: res.EnergyLimit, + EnergyUsed: res.EnergyUsed, + } + for _, d := range dels { + est := int64(float64(d.FrozenBalanceSun) * energyPerSun) + snap.TotalInboundEnergy += est + if d.ExpireTimeMs > cut24 { + snap.Protected24H += est + } + if d.ExpireTimeMs > cutLook { + snap.ProtectedLookahead += est + } + if d.ExpireTimeMs > cut7d { + snap.Protected7D += est + } + if d.ExpireTimeMs > nowMs && d.ExpireTimeMs <= cutLook { + snap.ExpiringCount++ + snap.ExpiringEnergy += est + } + if d.ExpireTimeMs > nowMs && (snap.NearestExpiryMs == 0 || d.ExpireTimeMs < snap.NearestExpiryMs) { + snap.NearestExpiryMs = d.ExpireTimeMs + } + } + return snap, nil +} diff --git a/chains/tron/energy_calc_test.go b/chains/tron/energy_calc_test.go new file mode 100644 index 0000000..7c9fcf6 --- /dev/null +++ b/chains/tron/energy_calc_test.go @@ -0,0 +1,65 @@ +package tron + +import ( + "testing" + "time" +) + +func TestComputeEnergySnapshot(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + lookahead := 72 * time.Hour + hourMs := int64(3600 * 1000) + // energyPerTRX = TotalEnergyLimit/TotalEnergyWeight = 180e9/6e9 = 30 + res := ResourceParams{ + EnergyLimit: 500_000, + EnergyUsed: 120_000, + TotalEnergyLimit: 180_000_000_000, + TotalEnergyWeight: 6_000_000_000, + } + // 1000 TRX = 1_000_000_000 sun → 30_000 energy + trx1000 := int64(1_000_000_000) + + dels := []DelegationDetail{ + {From: "A", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() + 200*hourMs}, // 8天后到期：全部窗口受保护 + {From: "B", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() + 48*hourMs}, // 48h 后到期：只受 24h 保护 + {From: "C", FrozenBalanceSun: trx1000, ExpireTimeMs: now.UnixMilli() - hourMs}, // 已到期 + {From: "D", FrozenBalanceSun: trx1000, ExpireTimeMs: 0}, // 无锁定 + {From: "E", FrozenBalanceSun: trx1000, ExpireTimeMs: now.Add(lookahead).UnixMilli()}, // 恰好=cutoff：不计入（严格晚于） + } + + snap, err := ComputeEnergySnapshot("T1", dels, res, now, lookahead) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if snap.TotalInboundEnergy != 150_000 { + t.Errorf("TotalInboundEnergy=%d, want 150000", snap.TotalInboundEnergy) + } + if snap.Protected24H != 90_000 { // A + B + E（E 到期于 72h > 24h 窗口） + t.Errorf("Protected24H=%d, want 90000", snap.Protected24H) + } + if snap.ProtectedLookahead != 30_000 { // 仅 A + t.Errorf("ProtectedLookahead=%d, want 30000", snap.ProtectedLookahead) + } + if snap.Protected7D != 30_000 { // 仅 A（200h > 168h） + t.Errorf("Protected7D=%d, want 30000", snap.Protected7D) + } + // lookahead 窗口内到期：B(48h) 和 E(恰好 72h)，C 已到期、D 无锁定不算 + if snap.ExpiringCount != 2 || snap.ExpiringEnergy != 60_000 { + t.Errorf("Expiring=%d/%d, want 2/60000", snap.ExpiringCount, snap.ExpiringEnergy) + } + if snap.NearestExpiryMs != now.UnixMilli()+48*hourMs { + t.Errorf("NearestExpiryMs=%d, want %d", snap.NearestExpiryMs, now.UnixMilli()+48*hourMs) + } + if snap.EnergyLimit != 500_000 || snap.EnergyUsed != 120_000 { + t.Errorf("limit/used=%d/%d, want 500000/120000", snap.EnergyLimit, snap.EnergyUsed) + } +} + +func TestComputeEnergySnapshotZeroWeight(t *testing.T) { + _, err := ComputeEnergySnapshot("T1", nil, + ResourceParams{TotalEnergyLimit: 1, TotalEnergyWeight: 0}, + time.UnixMilli(0), 72*time.Hour) + if err == nil { + t.Fatal("expected error on TotalEnergyWeight=0, got nil") + } +} From 7840e29cdbb4ba68263eba696e7c09372c1ebd53 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:09:53 +0800 Subject: [PATCH 05/13] feat(tron): add energy expiry alert state machine Co-Authored-By: Claude Fable 5 --- chains/tron/energy_state.go | 95 +++++++++++++++++++++ chains/tron/energy_state_test.go | 140 +++++++++++++++++++++++++++++++ 2 files changed, 235 insertions(+) create mode 100644 chains/tron/energy_state.go create mode 100644 chains/tron/energy_state_test.go diff --git a/chains/tron/energy_state.go b/chains/tron/energy_state.go new file mode 100644 index 0000000..6bf5931 --- /dev/null +++ b/chains/tron/energy_state.go @@ -0,0 +1,95 @@ +package tron + +import "time" + +type AlertStatus string + +const ( + StatusOK AlertStatus = "OK" + StatusAlert AlertStatus = "ALERT" + StatusUnknown AlertStatus = "UNKNOWN" +) + +type AlertAction int + +const ( + ActionNone AlertAction = iota + ActionFirstAlert + ActionRepeatAlert + ActionEscalateAlert + ActionRecovery + ActionFailureAlert +) + +// ExpiryPolicy is the per-address alerting policy (already defaulted). +type ExpiryPolicy struct { + ProtectedThreshold int64 + RecoveryThreshold int64 + RepeatInterval time.Duration + FailureThreshold int +} + +// AlertState is the persisted alert-dedup state for one address. +// Status only holds the last confirmed OK/ALERT baseline; scan failures are +// tracked via ConsecutiveFails and never overwrite the baseline, so an API +// outage can neither trigger nor clear an energy alert. +type AlertState struct { + Status AlertStatus `json:"status,omitempty"` + FirstAlertAtMs int64 `json:"firstAlertAtMs,omitempty"` + LastSentAtMs int64 `json:"lastSentAtMs,omitempty"` + LastRecoveredAtMs int64 `json:"lastRecoveredAtMs,omitempty"` + LastSentProtected int64 `json:"lastSentProtected,omitempty"` + ConsecutiveFails int `json:"consecutiveFails,omitempty"` + FailureAlerted bool `json:"failureAlerted,omitempty"` +} + +// NextAlertState advances the state machine for one scan result and returns +// the action to perform. snap may be nil when scanFailed is true. +func NextAlertState(st AlertState, snap *EnergySnapshot, scanFailed bool, pol ExpiryPolicy, now time.Time) (AlertState, AlertAction) { + if scanFailed { + st.ConsecutiveFails++ + if st.ConsecutiveFails >= pol.FailureThreshold && !st.FailureAlerted { + st.FailureAlerted = true + return st, ActionFailureAlert + } + return st, ActionNone + } + + st.ConsecutiveFails = 0 + st.FailureAlerted = false + nowMs := now.UnixMilli() + + if snap.ProtectedLookahead < pol.ProtectedThreshold { + if st.Status != StatusAlert { // first alert (baseline was OK/empty) + st.Status = StatusAlert + st.FirstAlertAtMs = nowMs + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionFirstAlert + } + prevGap := pol.ProtectedThreshold - st.LastSentProtected + curGap := pol.ProtectedThreshold - snap.ProtectedLookahead + if prevGap > 0 && float64(curGap) > float64(prevGap)*1.2 { + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionEscalateAlert + } + if nowMs-st.LastSentAtMs >= pol.RepeatInterval.Milliseconds() { + st.LastSentAtMs = nowMs + st.LastSentProtected = snap.ProtectedLookahead + return st, ActionRepeatAlert + } + return st, ActionNone + } + + if st.Status == StatusAlert { + if snap.ProtectedLookahead >= pol.RecoveryThreshold { + st.Status = StatusOK + st.LastRecoveredAtMs = nowMs + return st, ActionRecovery + } + return st, ActionNone // recovery buffer: stay ALERT silently + } + st.Status = StatusOK + return st, ActionNone +} diff --git a/chains/tron/energy_state_test.go b/chains/tron/energy_state_test.go new file mode 100644 index 0000000..4e3d0ac --- /dev/null +++ b/chains/tron/energy_state_test.go @@ -0,0 +1,140 @@ +package tron + +import ( + "testing" + "time" +) + +func pol() ExpiryPolicy { + return ExpiryPolicy{ + ProtectedThreshold: 10_000_000, + RecoveryThreshold: 10_500_000, + RepeatInterval: 12 * time.Hour, + FailureThreshold: 3, + } +} + +func snapWith(protected int64) *EnergySnapshot { + return &EnergySnapshot{ProtectedLookahead: protected} +} + +func TestNextAlertState(t *testing.T) { + base := time.UnixMilli(1_700_000_000_000) + + t.Run("first alert on drop below threshold", func(t *testing.T) { + st, act := NextAlertState(AlertState{}, snapWith(8_000_000), false, pol(), base) + if act != ActionFirstAlert { + t.Fatalf("action=%v, want ActionFirstAlert", act) + } + if st.Status != StatusAlert || st.FirstAlertAtMs != base.UnixMilli() || + st.LastSentAtMs != base.UnixMilli() || st.LastSentProtected != 8_000_000 { + t.Fatalf("state=%+v", st) + } + }) + + t.Run("no repeat within interval", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + _, act := NextAlertState(prev, snapWith(8_100_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone", act) + } + }) + + t.Run("repeat after interval", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(8_000_000), false, pol(), base.Add(12*time.Hour)) + if act != ActionRepeatAlert { + t.Fatalf("action=%v, want ActionRepeatAlert", act) + } + if st.LastSentAtMs != base.Add(12*time.Hour).UnixMilli() { + t.Fatalf("LastSentAtMs not updated: %+v", st) + } + }) + + t.Run("escalate when gap grows over 20 percent", func(t *testing.T) { + // prev gap = 10M-8M = 2M；current 7.5M → gap 2.5M = +25% + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(7_500_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionEscalateAlert { + t.Fatalf("action=%v, want ActionEscalateAlert", act) + } + if st.LastSentProtected != 7_500_000 { + t.Fatalf("LastSentProtected not updated: %+v", st) + } + }) + + t.Run("gap growth under 20 percent does not escalate", func(t *testing.T) { + // prev gap 2M；current 7.7M → gap 2.3M = +15% + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + _, act := NextAlertState(prev, snapWith(7_700_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone", act) + } + }) + + t.Run("recovery buffer holds alert", func(t *testing.T) { + // 10.2M：≥阈值但 <恢复阈值 → 维持 ALERT 不发 + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(10_200_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionNone || st.Status != StatusAlert { + t.Fatalf("action=%v status=%v, want ActionNone/ALERT", act, st.Status) + } + }) + + t.Run("recovery above recovery threshold", func(t *testing.T) { + prev := AlertState{Status: StatusAlert, LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(prev, snapWith(10_600_000), false, pol(), base.Add(1*time.Hour)) + if act != ActionRecovery || st.Status != StatusOK { + t.Fatalf("action=%v status=%v, want ActionRecovery/OK", act, st.Status) + } + if st.LastRecoveredAtMs != base.Add(1*time.Hour).UnixMilli() { + t.Fatalf("LastRecoveredAtMs not set: %+v", st) + } + }) + + t.Run("ok stays ok", func(t *testing.T) { + st, act := NextAlertState(AlertState{}, snapWith(20_000_000), false, pol(), base) + if act != ActionNone || st.Status != StatusOK { + t.Fatalf("action=%v status=%v, want ActionNone/OK", act, st.Status) + } + }) + + t.Run("scan failure never triggers energy alert", func(t *testing.T) { + st, act := NextAlertState(AlertState{Status: StatusOK}, nil, true, pol(), base) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone on 1st failure", act) + } + if st.ConsecutiveFails != 1 || st.Status != StatusOK { + t.Fatalf("state=%+v, want fails=1 and baseline kept", st) + } + }) + + t.Run("failure alert on 3rd consecutive failure, only once", func(t *testing.T) { + st := AlertState{Status: StatusAlert, ConsecutiveFails: 2} + st, act := NextAlertState(st, nil, true, pol(), base) + if act != ActionFailureAlert || !st.FailureAlerted { + t.Fatalf("action=%v state=%+v, want ActionFailureAlert", act, st) + } + if st.Status != StatusAlert { + t.Fatalf("baseline lost: %+v", st) + } + _, act = NextAlertState(st, nil, true, pol(), base.Add(time.Hour)) + if act != ActionNone { + t.Fatalf("action=%v, want ActionNone on 4th failure (already alerted)", act) + } + }) + + t.Run("success after unknown resets fails and rejudges against baseline", func(t *testing.T) { + // 之前是 ALERT，UNKNOWN 若干轮后恢复查询且低于阈值 → 不是首报（基线还是 ALERT）， + // 距上次发送超 12h → 重复提醒 + st := AlertState{Status: StatusAlert, ConsecutiveFails: 5, FailureAlerted: true, + LastSentAtMs: base.UnixMilli(), LastSentProtected: 8_000_000} + st, act := NextAlertState(st, snapWith(8_000_000), false, pol(), base.Add(13*time.Hour)) + if act != ActionRepeatAlert { + t.Fatalf("action=%v, want ActionRepeatAlert", act) + } + if st.ConsecutiveFails != 0 || st.FailureAlerted { + t.Fatalf("failure tracking not reset: %+v", st) + } + }) +} From 6b60428392db420d67d8bc91d4a5c7776dc5d2a0 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:10:39 +0800 Subject: [PATCH 06/13] feat(tron): add JSON persistence for energy expiry state Co-Authored-By: Claude Fable 5 --- chains/tron/energy_store.go | 71 ++++++++++++++++++++++++++++++++ chains/tron/energy_store_test.go | 60 +++++++++++++++++++++++++++ 2 files changed, 131 insertions(+) create mode 100644 chains/tron/energy_store.go create mode 100644 chains/tron/energy_store_test.go diff --git a/chains/tron/energy_store.go b/chains/tron/energy_store.go new file mode 100644 index 0000000..cbb8edb --- /dev/null +++ b/chains/tron/energy_store.go @@ -0,0 +1,71 @@ +package tron + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +// maxSnapshotHistory bounds the persisted scan history: 168 hourly scans ≈ 7 days. +const maxSnapshotHistory = 168 + +// SnapshotEntry is one scan result kept for audit. +type SnapshotEntry struct { + AtMs int64 `json:"atMs"` + Status AlertStatus `json:"status"` + Err string `json:"err,omitempty"` + Snap *EnergySnapshot `json:"snapshot,omitempty"` +} + +// StoredEnergyState is the on-disk state for one monitored address. +type StoredEnergyState struct { + Address string `json:"address"` + Alert AlertState `json:"alert"` + Snapshots []SnapshotEntry `json:"snapshots"` +} + +// Append adds an entry, dropping the oldest beyond maxSnapshotHistory. +func (s *StoredEnergyState) Append(e SnapshotEntry) { + s.Snapshots = append(s.Snapshots, e) + if n := len(s.Snapshots) - maxSnapshotHistory; n > 0 { + s.Snapshots = append(s.Snapshots[:0:0], s.Snapshots[n:]...) + } +} + +// LoadEnergyState reads state from path. A missing file yields a fresh state +// and nil error; a corrupt file yields a fresh state AND the error, so the +// caller can log/alarm but keep monitoring. +func LoadEnergyState(path, address string) (*StoredEnergyState, error) { + fresh := &StoredEnergyState{Address: address} + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return fresh, nil + } + return fresh, err + } + st := &StoredEnergyState{} + if err := json.Unmarshal(data, st); err != nil { + return fresh, fmt.Errorf("corrupt energy state %s: %w", path, err) + } + st.Address = address + return st, nil +} + +// SaveEnergyState writes state atomically (tmp file + rename), creating the +// parent directory when needed. +func SaveEnergyState(path string, st *StoredEnergyState) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + data, err := json.MarshalIndent(st, "", " ") + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} diff --git a/chains/tron/energy_store_test.go b/chains/tron/energy_store_test.go new file mode 100644 index 0000000..2c3e975 --- /dev/null +++ b/chains/tron/energy_store_test.go @@ -0,0 +1,60 @@ +package tron + +import ( + "os" + "path/filepath" + "testing" +) + +func TestEnergyStateRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "sub", "energy_state_T1.json") + + st, err := LoadEnergyState(path, "T1") + if err != nil { + t.Fatalf("load missing file: %v", err) + } + if st.Address != "T1" || len(st.Snapshots) != 0 { + t.Fatalf("fresh state=%+v", st) + } + + st.Alert = AlertState{Status: StatusAlert, LastSentProtected: 123} + st.Append(SnapshotEntry{AtMs: 1, Status: StatusAlert, Snap: &EnergySnapshot{Address: "T1"}}) + if err := SaveEnergyState(path, st); err != nil { + t.Fatalf("save: %v", err) + } + + got, err := LoadEnergyState(path, "T1") + if err != nil { + t.Fatalf("reload: %v", err) + } + if got.Alert.Status != StatusAlert || got.Alert.LastSentProtected != 123 || len(got.Snapshots) != 1 { + t.Fatalf("reloaded=%+v", got) + } +} + +func TestEnergyStateHistoryCap(t *testing.T) { + st := &StoredEnergyState{Address: "T1"} + for i := 0; i < maxSnapshotHistory+10; i++ { + st.Append(SnapshotEntry{AtMs: int64(i)}) + } + if len(st.Snapshots) != maxSnapshotHistory { + t.Fatalf("len=%d, want %d", len(st.Snapshots), maxSnapshotHistory) + } + if st.Snapshots[0].AtMs != 10 { + t.Fatalf("oldest=%d, want 10 (ring dropped head)", st.Snapshots[0].AtMs) + } +} + +func TestLoadEnergyStateCorrupt(t *testing.T) { + path := filepath.Join(t.TempDir(), "energy_state_T1.json") + if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil { + t.Fatal(err) + } + st, err := LoadEnergyState(path, "T1") + if err == nil { + t.Fatal("expected error for corrupt file") + } + if st == nil || st.Address != "T1" { + t.Fatalf("must still return usable fresh state, got %+v", st) + } +} From 61285381ea2a7524a06e545712d810788e136017 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:11:46 +0800 Subject: [PATCH 07/13] feat(tron): add inbound delegation gRPC queries with retry Co-Authored-By: Claude Fable 5 --- chains/tron/con.go | 116 +++++++++++++++++++++++++++++++++++++++- chains/tron/con_test.go | 34 ++++++++++++ 2 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 chains/tron/con_test.go diff --git a/chains/tron/con.go b/chains/tron/con.go index 5e04cf8..cd08d43 100644 --- a/chains/tron/con.go +++ b/chains/tron/con.go @@ -1,10 +1,12 @@ package tron import ( - "github.com/ethereum/go-ethereum/ethclient" + "context" "math/big" "time" + "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/accounts/keystore" "google.golang.org/grpc" @@ -12,6 +14,10 @@ import ( "github.com/ethereum/go-ethereum/accounts/abi/bind" ethcommon "github.com/ethereum/go-ethereum/common" "github.com/lbtsm/gotron-sdk/pkg/client" + "github.com/lbtsm/gotron-sdk/pkg/common" + "github.com/lbtsm/gotron-sdk/pkg/proto/api" + "github.com/lbtsm/gotron-sdk/pkg/proto/core" + "github.com/pkg/errors" ) type Connection struct { @@ -96,3 +102,111 @@ func (c *Connection) Close() { } close(c.stop) } + +const ( + delegationRequestGap = 500 * time.Millisecond + queryMaxRetries = 3 + queryRetryBackoff = time.Second + queryTimeout = 15 * time.Second +) + +// withRetry runs do up to attempts times with exponential backoff. +func withRetry(attempts int, backoff time.Duration, do func() error) error { + var err error + for i := 0; i < attempts; i++ { + if err = do(); err == nil { + return nil + } + if i < attempts-1 { + time.Sleep(backoff) + backoff *= 2 + } + } + return err +} + +// InboundDelegations lists every energy delegation TO target (Stake 2.0). +// Any final failure returns an error — callers must treat the whole scan as +// UNKNOWN rather than compute from partial data. +// +// Note: the SDK's GetDelegatedResourcesV2 walks ToAccounts (outbound), which +// is the opposite direction, so we drive the raw stubs ourselves. +func (c *Connection) InboundDelegations(target string) ([]DelegationDetail, error) { + targetBytes, err := common.DecodeCheck(target) + if err != nil { + return nil, errors.Wrapf(err, "decode address %s", target) + } + + index, err := c.delegationIndex(targetBytes) + if err != nil { + return nil, errors.Wrap(err, "GetDelegatedResourceAccountIndexV2") + } + + details := make([]DelegationDetail, 0, len(index.GetFromAccounts())) + for i, from := range index.GetFromAccounts() { + if i > 0 { + time.Sleep(delegationRequestGap) + } + list, err := c.delegationDetail(from, targetBytes) + if err != nil { + return nil, errors.Wrapf(err, "GetDelegatedResourceV2 from %s", common.EncodeCheck(from)) + } + for _, d := range list.GetDelegatedResource() { + if d.GetFrozenBalanceForEnergy() <= 0 { + continue // bandwidth-only delegation + } + details = append(details, DelegationDetail{ + From: common.EncodeCheck(d.GetFrom()), + FrozenBalanceSun: d.GetFrozenBalanceForEnergy(), + ExpireTimeMs: d.GetExpireTimeForEnergy(), + }) + } + } + return details, nil +} + +func (c *Connection) delegationIndex(target []byte) (*core.DelegatedResourceAccountIndex, error) { + var index *core.DelegatedResourceAccountIndex + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + ctx, cancel := context.WithTimeout(context.Background(), queryTimeout) + defer cancel() + var e error + index, e = c.cli.Client.GetDelegatedResourceAccountIndexV2(ctx, client.GetMessageBytes(target)) + return e + }) + return index, err +} + +func (c *Connection) delegationDetail(from, to []byte) (*api.DelegatedResourceList, error) { + var list *api.DelegatedResourceList + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + ctx, cancel := context.WithTimeout(context.Background(), queryTimeout) + defer cancel() + var e error + list, e = c.cli.Client.GetDelegatedResourceV2(ctx, &api.DelegatedResourceMessage{ + FromAddress: from, + ToAddress: to, + }) + return e + }) + return list, err +} + +// EnergyResourceParams reads the numbers needed for sun→energy conversion. +func (c *Connection) EnergyResourceParams(target string) (ResourceParams, error) { + var params ResourceParams + err := withRetry(queryMaxRetries, queryRetryBackoff, func() error { + res, e := c.cli.GetAccountResource(target) + if e != nil { + return e + } + params = ResourceParams{ + EnergyLimit: res.GetEnergyLimit(), + EnergyUsed: res.GetEnergyUsed(), + TotalEnergyLimit: res.GetTotalEnergyLimit(), + TotalEnergyWeight: res.GetTotalEnergyWeight(), + } + return nil + }) + return params, err +} diff --git a/chains/tron/con_test.go b/chains/tron/con_test.go new file mode 100644 index 0000000..f77ff3a --- /dev/null +++ b/chains/tron/con_test.go @@ -0,0 +1,34 @@ +package tron + +import ( + "errors" + "testing" + "time" +) + +func TestWithRetry(t *testing.T) { + t.Run("succeeds after transient failures", func(t *testing.T) { + calls := 0 + err := withRetry(3, time.Millisecond, func() error { + calls++ + if calls < 3 { + return errors.New("transient") + } + return nil + }) + if err != nil || calls != 3 { + t.Fatalf("err=%v calls=%d, want nil/3", err, calls) + } + }) + + t.Run("returns last error when exhausted", func(t *testing.T) { + calls := 0 + err := withRetry(3, time.Millisecond, func() error { + calls++ + return errors.New("boom") + }) + if err == nil || calls != 3 { + t.Fatalf("err=%v calls=%d, want error/3", err, calls) + } + }) +} From 307939f54cff2ab97b26b4f403efcb38bf392d2b Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:13:34 +0800 Subject: [PATCH 08/13] feat(tron): add energy expiry checker with dedup alerting Co-Authored-By: Claude Fable 5 --- chains/tron/energy.go | 197 +++++++++++++++++++++++++++++++++++++ chains/tron/energy_test.go | 130 ++++++++++++++++++++++++ 2 files changed, 327 insertions(+) create mode 100644 chains/tron/energy.go create mode 100644 chains/tron/energy_test.go diff --git a/chains/tron/energy.go b/chains/tron/energy.go new file mode 100644 index 0000000..fb39a0c --- /dev/null +++ b/chains/tron/energy.go @@ -0,0 +1,197 @@ +package tron + +import ( + "context" + "fmt" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/ChainSafe/log15" + "github.com/mapprotocol/monitor/internal/config" +) + +// energyQuerier is the chain-access surface the checker needs; *Connection +// implements it, tests use a fake. +type energyQuerier interface { + InboundDelegations(target string) ([]DelegationDetail, error) + EnergyResourceParams(target string) (ResourceParams, error) +} + +const energyFailureThreshold = 3 + +// displayLoc is the timezone used in alert messages (storage stays UTC ms). +var displayLoc = func() *time.Location { + loc, err := time.LoadLocation("Asia/Singapore") + if err != nil { + return time.UTC + } + return loc +}() + +// expiryChecker runs protected-energy scans and drives alerting for the +// addresses configured with a protectedThreshold. +type expiryChecker struct { + log log15.Logger + q energyQuerier + chain string + dir string // state-file directory + states map[string]*StoredEnergyState + alarm func(ctx context.Context, msg string) +} + +func newExpiryChecker(log log15.Logger, q energyQuerier, chainName, dir string, + alarm func(ctx context.Context, msg string)) *expiryChecker { + return &expiryChecker{ + log: log, + q: q, + chain: chainName, + dir: dir, + states: make(map[string]*StoredEnergyState), + alarm: alarm, + } +} + +func (c *expiryChecker) statePath(addr string) string { + return filepath.Join(c.dir, "energy_state_"+addr+".json") +} + +func (c *expiryChecker) state(addr string) *StoredEnergyState { + if st, ok := c.states[addr]; ok { + return st + } + st, err := LoadEnergyState(c.statePath(addr), addr) + if err != nil { + c.log.Error("EnergyExpiry state file unreadable, starting fresh", "addr", addr, "err", err) + } + c.states[addr] = st + return st +} + +// runOnce performs one scan+judge+persist+alert cycle for one address. +// en must already have ApplyExpiryDefaults applied. +func (c *expiryChecker) runOnce(en config.Energy, now time.Time) { + st := c.state(en.Address) + pol := ExpiryPolicy{ + ProtectedThreshold: en.ProtectedThreshold, + RecoveryThreshold: en.RecoveryThreshold, + RepeatInterval: time.Duration(en.RepeatIntervalHours) * time.Hour, + FailureThreshold: energyFailureThreshold, + } + lookahead := time.Duration(en.LookaheadHours) * time.Hour + + snap, scanErr := c.scan(en.Address, now, lookahead) + newAlert, action := NextAlertState(st.Alert, snap, scanErr != nil, pol, now) + + entry := SnapshotEntry{AtMs: now.UnixMilli(), Snap: snap} + if scanErr != nil { + entry.Status = StatusUnknown + entry.Err = scanErr.Error() + c.log.Error("EnergyExpiry scan failed", "addr", en.Address, + "consecutiveFails", newAlert.ConsecutiveFails, "err", scanErr) + } else { + entry.Status = newAlert.Status + c.log.Info("EnergyExpiry scan", "addr", en.Address, + "protected", snap.ProtectedLookahead, "threshold", en.ProtectedThreshold, + "expiring", snap.ExpiringCount, "status", newAlert.Status) + } + st.Alert = newAlert + st.Append(entry) + if err := SaveEnergyState(c.statePath(en.Address), st); err != nil { + c.log.Error("EnergyExpiry save state failed", "addr", en.Address, "err", err) + } + + switch action { + case ActionFirstAlert, ActionRepeatAlert, ActionEscalateAlert: + c.alarm(context.Background(), formatEnergyAlert(c.chain, en, snap, now, lookahead)) + case ActionRecovery: + c.alarm(context.Background(), formatEnergyRecovery(c.chain, en, snap)) + case ActionFailureAlert: + c.alarm(context.Background(), formatEnergyFailure(c.chain, en.Address, newAlert.ConsecutiveFails, scanErr)) + } +} + +func (c *expiryChecker) scan(addr string, now time.Time, lookahead time.Duration) (*EnergySnapshot, error) { + dels, err := c.q.InboundDelegations(addr) + if err != nil { + return nil, err + } + res, err := c.q.EnergyResourceParams(addr) + if err != nil { + return nil, err + } + return ComputeEnergySnapshot(addr, dels, res, now, lookahead) +} + +func fmtEnergyTime(ms int64) string { + if ms <= 0 { + return "-" + } + return time.UnixMilli(ms).In(displayLoc).Format("2006-01-02 15:04 MST") +} + +func fmtEnergyInt(n int64) string { + neg := n < 0 + if neg { + n = -n + } + s := strconv.FormatInt(n, 10) + var b strings.Builder + pre := len(s) % 3 + if pre > 0 { + b.WriteString(s[:pre]) + } + for i := pre; i < len(s); i += 3 { + if b.Len() > 0 { + b.WriteByte(',') + } + b.WriteString(s[i : i+3]) + } + if neg { + return "-" + b.String() + } + return b.String() +} + +func formatEnergyAlert(chainName string, en config.Energy, snap *EnergySnapshot, now time.Time, lookahead time.Duration) string { + return fmt.Sprintf(`[TRON Energy 预警] +链：%s +地址：%s +预测时间：%s +%d 小时后受保护 Energy：%s +配置阈值：%s +缺口：%s +未来 %d 小时到期：%d 笔，共约 %s Energy +最近到期：%s +查询时间：%s`, + chainName, en.Address, + fmtEnergyTime(now.Add(lookahead).UnixMilli()), + en.LookaheadHours, fmtEnergyInt(snap.ProtectedLookahead), + fmtEnergyInt(en.ProtectedThreshold), + fmtEnergyInt(en.ProtectedThreshold-snap.ProtectedLookahead), + en.LookaheadHours, snap.ExpiringCount, fmtEnergyInt(snap.ExpiringEnergy), + fmtEnergyTime(snap.NearestExpiryMs), + fmtEnergyTime(snap.QueriedAtMs)) +} + +func formatEnergyRecovery(chainName string, en config.Energy, snap *EnergySnapshot) string { + return fmt.Sprintf(`[TRON Energy 恢复] +链：%s +地址：%s +%d 小时后受保护 Energy：%s（≥ 恢复阈值 %s） +查询时间：%s`, + chainName, en.Address, + en.LookaheadHours, fmtEnergyInt(snap.ProtectedLookahead), + fmtEnergyInt(en.RecoveryThreshold), + fmtEnergyTime(snap.QueriedAtMs)) +} + +func formatEnergyFailure(chainName, addr string, fails int, scanErr error) string { + return fmt.Sprintf(`[TRON Energy 监控异常] +链：%s +地址：%s +连续 %d 次查询失败，受保护 Energy 状态未知（不代表 Energy 为 0） +最近错误：%v`, + chainName, addr, fails, scanErr) +} diff --git a/chains/tron/energy_test.go b/chains/tron/energy_test.go new file mode 100644 index 0000000..b9d6f10 --- /dev/null +++ b/chains/tron/energy_test.go @@ -0,0 +1,130 @@ +package tron + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/ChainSafe/log15" + "github.com/mapprotocol/monitor/internal/config" +) + +type fakeQuerier struct { + dels []DelegationDetail + res ResourceParams + err error +} + +func (f *fakeQuerier) InboundDelegations(string) ([]DelegationDetail, error) { + if f.err != nil { + return nil, f.err + } + return f.dels, nil +} + +func (f *fakeQuerier) EnergyResourceParams(string) (ResourceParams, error) { + if f.err != nil { + return ResourceParams{}, f.err + } + return f.res, nil +} + +func testEnergyCfg() config.Energy { + en := config.Energy{Address: "TTest1", ProtectedThreshold: 100_000} + en.ApplyExpiryDefaults() + return en +} + +func newTestChecker(t *testing.T, q energyQuerier) (*expiryChecker, *[]string) { + t.Helper() + var sent []string + c := newExpiryChecker(log15.New(), q, "tron", t.TempDir(), + func(_ context.Context, msg string) { sent = append(sent, msg) }) + return c, &sent +} + +func TestRunOnceFirstAlertAndPersist(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + q := &fakeQuerier{ + // 30 TRX 无锁定 → protected 0 < 100_000 阈值 + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 30_000_000, ExpireTimeMs: 0}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c, sent := newTestChecker(t, q) + en := testEnergyCfg() + + c.runOnce(en, now) + + if len(*sent) != 1 || !strings.Contains((*sent)[0], "TRON Energy 预警") { + t.Fatalf("sent=%v, want one first alert", *sent) + } + // 状态已持久化，重启（新 checker）后同样低于阈值不再首报 + c2, sent2 := newTestChecker(t, q) + c2.dir = c.dir + c2.runOnce(en, now.Add(time.Hour)) + if len(*sent2) != 0 { + t.Fatalf("sent after restart=%v, want dedup (no resend within 12h)", *sent2) + } + // 两轮扫描（首报一轮 + 重启后去重一轮）都应留下快照记录 + st, err := LoadEnergyState(filepath.Join(c.dir, "energy_state_TTest1.json"), "TTest1") + if err != nil || st.Alert.Status != StatusAlert || len(st.Snapshots) != 2 { + t.Fatalf("persisted state=%+v err=%v", st, err) + } +} + +func TestRunOnceFailureIsUnknownNotZero(t *testing.T) { + q := &fakeQuerier{err: errors.New("rpc down")} + c, sent := newTestChecker(t, q) + en := testEnergyCfg() + now := time.UnixMilli(1_700_000_000_000) + + c.runOnce(en, now) + c.runOnce(en, now.Add(time.Hour)) + if len(*sent) != 0 { + t.Fatalf("sent=%v, want none before 3rd failure", *sent) + } + c.runOnce(en, now.Add(2*time.Hour)) + if len(*sent) != 1 || !strings.Contains((*sent)[0], "监控异常") { + t.Fatalf("sent=%v, want one failure alert", *sent) + } + if strings.Contains((*sent)[0], "预警]") { + t.Fatalf("failure alert must not look like an energy alert: %v", *sent) + } +} + +func TestRunOnceRecovery(t *testing.T) { + now := time.UnixMilli(1_700_000_000_000) + low := &fakeQuerier{ + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 30_000_000, ExpireTimeMs: 0}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c, sent := newTestChecker(t, low) + en := testEnergyCfg() + c.runOnce(en, now) // first alert + + // 换成充足数据：10000 TRX 锁到 30 天后 → 300_000 > recovery 105_000 + c.q = &fakeQuerier{ + dels: []DelegationDetail{{From: "A", FrozenBalanceSun: 10_000_000_000, + ExpireTimeMs: now.Add(30 * 24 * time.Hour).UnixMilli()}}, + res: ResourceParams{TotalEnergyLimit: 180_000_000_000, TotalEnergyWeight: 6_000_000_000}, + } + c.runOnce(en, now.Add(time.Hour)) + if len(*sent) != 2 || !strings.Contains((*sent)[1], "恢复") { + t.Fatalf("sent=%v, want alert then recovery", *sent) + } +} + +func TestFmtEnergyInt(t *testing.T) { + tests := []struct { + in int64 + want string + }{{0, "0"}, {999, "999"}, {1000, "1,000"}, {10500000, "10,500,000"}, {-1234, "-1,234"}} + for _, tt := range tests { + if got := fmtEnergyInt(tt.in); got != tt.want { + t.Errorf("fmtEnergyInt(%d)=%q, want %q", tt.in, got, tt.want) + } + } +} From 724751c60350b5e0743c56f37498c6272aa23c74 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:14:26 +0800 Subject: [PATCH 09/13] feat(tron): wire energy expiry checker into monitor sync Co-Authored-By: Claude Fable 5 --- chains/tron/energy.go | 38 ++++++++++++++++++++++++++++++++++++++ chains/tron/monitor.go | 8 ++++++++ 2 files changed, 46 insertions(+) diff --git a/chains/tron/energy.go b/chains/tron/energy.go index fb39a0c..023ba39 100644 --- a/chains/tron/energy.go +++ b/chains/tron/energy.go @@ -9,7 +9,10 @@ import ( "time" "github.com/ChainSafe/log15" + "github.com/mapprotocol/monitor/internal/chain" "github.com/mapprotocol/monitor/internal/config" + "github.com/mapprotocol/monitor/pkg/util" + "github.com/pkg/errors" ) // energyQuerier is the chain-access surface the checker needs; *Connection @@ -187,6 +190,41 @@ func formatEnergyRecovery(chainName string, en config.Energy, snap *EnergySnapsh fmtEnergyTime(snap.QueriedAtMs)) } +// energyExpiryTick is how often the scheduler wakes to check per-address due +// times; actual scan cadence is each address's CheckIntervalMinutes. +const energyExpiryTick = time.Minute + +// energyExpirySync schedules expiry scans for all enabled energy entries. +// It re-reads the config snapshot every tick, so hot-reloaded thresholds and +// newly added addresses are picked up without restart. +func (m *Monitor) energyExpirySync() error { + checker := newExpiryChecker(m.Log, m.conn, m.Cfg.Name, m.Cfg.KeystorePath, util.Alarm) + nextRun := make(map[string]time.Time) + for { + select { + case <-m.Stop: + return errors.New("energy expiry polling terminated") + default: + snap := m.Snapshot() + for _, en := range snap.Energies { + if en.ProtectedThreshold <= 0 { + continue + } + en.ApplyExpiryDefaults() + now := time.Now() + if now.Before(nextRun[en.Address]) { + continue + } + checker.runOnce(en, now) + nextRun[en.Address] = now.Add(time.Duration(en.CheckIntervalMinutes) * time.Minute) + } + if !chain.SleepWithStop(m.Stop, energyExpiryTick) { + return errors.New("energy expiry polling terminated") + } + } + } +} + func formatEnergyFailure(chainName, addr string, fails int, scanErr error) string { return fmt.Sprintf(`[TRON Energy 监控异常] 链：%s diff --git a/chains/tron/monitor.go b/chains/tron/monitor.go index 30d6492..9d13b16 100644 --- a/chains/tron/monitor.go +++ b/chains/tron/monitor.go @@ -47,6 +47,14 @@ func (m *Monitor) Sync() error { } }() + m.Wg.Add(1) + go func() { + defer m.Wg.Done() + if err := m.energyExpirySync(); err != nil { + m.Log.Error("Energy expiry polling stopped", "err", err) + } + }() + return nil } From 0dc673a89490722beebf00fbf4c4a7a9c03a0ed5 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 14 Aug 2026 12:14:46 +0800 Subject: [PATCH 10/13] docs: document tron energy expiry monitoring config Co-Authored-By: Claude Fable 5 --- README.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/README.md b/README.md index 8bb742e..91dbb26 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,27 @@ See `config.example` for an example configuration. } ``` +## TRON Energy Expiry Monitoring + +Add `protectedThreshold` to a tron chain's `energy` entry to enable +Stake 2.0 delegation-expiry alerting (protected energy = inbound delegations +whose lock expires strictly after now+lookahead): + +```shell +"energy": [{ + "address": "TT6GDYkpHPVk24w9he9pavbagtzqBRS3XP", + "waterline": 100000, // existing: current remaining-energy alarm + "protectedThreshold": 10000000, // alert when protected energy drops below + "recoveryThreshold": 10500000, // optional, default = protected × 1.05 + "lookaheadHours": 72, // optional, default 72 + "checkIntervalMinutes": 60, // optional, default 60 + "repeatIntervalHours": 12 // optional, default 12 +}] +``` + +State files are written to `/energy_state_
.json`. +Scan failures alarm separately as "监控异常" and never count as zero energy. + ## Env ```shell From 48469b4550d8e7e915c888a4f6b5ddb290a59cd6 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Thu, 3 Sep 2026 16:52:52 +0800 Subject: [PATCH 11/13] Add Docker deployment and GitHub CD --- .dockerignore | 26 ++++++++++++ .env.example | 7 ++++ .github/workflows/cd.yml | 90 ++++++++++++++++++++++++++++++++++++++++ Dockerfile | 32 ++++++++++++++ README.md | 37 +++++++++++++++-- compose.yaml | 17 ++++++++ 6 files changed, 205 insertions(+), 4 deletions(-) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 .github/workflows/cd.yml create mode 100644 Dockerfile create mode 100644 compose.yaml diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..fa5a400 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,26 @@ +.git +.github +.codex +.agents +.DS_Store +.idea +.vscode +.history + +build +bin +site +target +tests/*.txt + +.env +.env.example +.config.example +config.json +config.example*.json +keys +*.key +*.block +gethdata1 +gethdata2 +centrifuge-chain diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..96091ad --- /dev/null +++ b/.env.example @@ -0,0 +1,7 @@ +MONITOR_DATA_DIR=/opt/bridge-monitor +MONITOR_IMAGE=bridge-monitor:local +MONITOR_CONTAINER_NAME=bridge-monitor +MONITOR_VERSION=local +TZ=Asia/Shanghai +compass=Mainnet-monitor +hooks= diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml new file mode 100644 index 0000000..9a58535 --- /dev/null +++ b/.github/workflows/cd.yml @@ -0,0 +1,90 @@ +name: CD + +on: + push: + branches: + - main + - master + tags: + - "v*" + pull_request: + workflow_dispatch: + +permissions: + contents: read + packages: write + +concurrency: + group: cd-${{ github.ref }} + cancel-in-progress: true + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + test: + name: Go test and build + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Go + uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + + - name: Run unit tests + run: go test ./internal/... ./chains/tron ./pkg/monitor + + - name: Build binary + run: go build -trimpath -ldflags="-s -w -X main.Version=${GITHUB_REF_NAME}" -o build/bridge-monitor ./cmd + + docker: + name: Docker build + runs-on: ubuntu-latest + needs: test + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Validate Compose file + run: docker compose config + + - name: Normalize image name + run: echo "IMAGE_NAME=${GITHUB_REPOSITORY,,}" >> "$GITHUB_ENV" + + - name: Docker metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=ref,event=branch + type=ref,event=tag + type=sha,prefix=sha- + + - name: Login to GitHub Container Registry + if: github.event_name != 'pull_request' + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Build and publish Docker image + uses: docker/build-push-action@v6 + with: + context: . + push: ${{ github.event_name != 'pull_request' }} + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + build-args: | + VERSION=${{ github.ref_name }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..fd23864 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,32 @@ +# syntax=docker/dockerfile:1 + +ARG GO_VERSION=1.25 + +FROM golang:${GO_VERSION}-bookworm AS builder + +WORKDIR /src + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +ARG VERSION=dev +RUN go build -trimpath -ldflags="-s -w -X main.Version=${VERSION}" -o /out/bridge-monitor ./cmd + +FROM debian:bookworm-slim AS runtime + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates tzdata \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app/runtime + +RUN mkdir -p /app/runtime/keys + +COPY --from=builder /out/bridge-monitor /usr/local/bin/bridge-monitor + +ENV TZ=Asia/Shanghai + +ENTRYPOINT ["bridge-monitor"] +CMD ["monitor", "--config", "/app/runtime/config.json"] diff --git a/README.md b/README.md index 91dbb26..7848be1 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,8 @@ whether the transaction is cross-chain, and the user balance # Configuration -See `config.example` for an example configuration. +Use a valid `config.json` for runtime configuration. Keep production config +files and keystores out of git. ## Options @@ -42,6 +43,34 @@ Scan failures alarm separately as "监控异常" and never count as zero energy. ## Env -```shell -export hooks="https://hooks.slack.com/services/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" // Slack alarm hook, Apply See This https://api.slack.com/messaging/webhooks -``` \ No newline at end of file +```shell +export hooks="https://hooks.slack.com/services/xxx/yyy/zzz" +``` + +# Docker Deployment + +The container uses `/app/runtime` as its runtime directory. Map one host +directory to it and keep `config.json`, `keys/`, and generated state files in +that host directory. + +```shell +sudo mkdir -p /opt/bridge-monitor/keys +sudo cp /path/to/your/config.json /opt/bridge-monitor/config.json + +cp .env.example .env +vim .env + +docker compose up -d --build +docker compose logs -f bridge-monitor +``` + +After pulling updates on the server: + +```shell +git pull +docker compose up -d --build +``` + +GitHub Actions builds and tests the project on pull requests, and builds the +Docker image on pushes to `main`, `master`, or version tags. Non-PR builds are +published to GitHub Container Registry as `ghcr.io//:`. diff --git a/compose.yaml b/compose.yaml new file mode 100644 index 0000000..04dee8c --- /dev/null +++ b/compose.yaml @@ -0,0 +1,17 @@ +services: + bridge-monitor: + build: + context: . + args: + VERSION: ${MONITOR_VERSION:-local} + image: ${MONITOR_IMAGE:-bridge-monitor:local} + container_name: ${MONITOR_CONTAINER_NAME:-bridge-monitor} + restart: unless-stopped + working_dir: /app/runtime + environment: + TZ: ${TZ:-Asia/Shanghai} + compass: ${compass:-} + hooks: ${hooks:-} + volumes: + - ${MONITOR_DATA_DIR:-/opt/bridge-monitor}:/app/runtime + command: ["monitor", "--config", "/app/runtime/config.json"] From c4238eefe5f8e888f73ce71edc20e9da99cbffe0 Mon Sep 17 00:00:00 2001 From: lbtsm Date: Fri, 4 Sep 2026 18:02:52 +0800 Subject: [PATCH 12/13] Support private Go modules in Docker CD --- .github/workflows/cd.yml | 17 +++++++++++++++++ Dockerfile | 15 ++++++++++++++- README.md | 12 ++++++++++++ compose.yaml | 6 ++++++ 4 files changed, 49 insertions(+), 1 deletion(-) diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml index 9a58535..0125922 100644 --- a/.github/workflows/cd.yml +++ b/.github/workflows/cd.yml @@ -21,6 +21,8 @@ concurrency: env: REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} + GOPRIVATE: github.com/lbtsm/* + GONOSUMDB: github.com/lbtsm/* jobs: test: @@ -36,6 +38,17 @@ jobs: go-version-file: go.mod cache: true + - name: Configure private Go module access + env: + LBTSM_REPO_TOKEN: ${{ secrets.LBTSM_REPO_TOKEN }} + run: | + if [ -z "$LBTSM_REPO_TOKEN" ]; then + echo "::warning::LBTSM_REPO_TOKEN is not set; private github.com/lbtsm modules must be public or Go module download will fail." + exit 0 + fi + printf "machine github.com\nlogin x-access-token\npassword %s\n" "$LBTSM_REPO_TOKEN" > ~/.netrc + chmod 0600 ~/.netrc + - name: Run unit tests run: go test ./internal/... ./chains/tron ./pkg/monitor @@ -65,6 +78,7 @@ jobs: with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} tags: | + type=raw,value=latest,enable={{is_default_branch}} type=ref,event=branch type=ref,event=tag type=sha,prefix=sha- @@ -86,5 +100,8 @@ jobs: labels: ${{ steps.meta.outputs.labels }} build-args: | VERSION=${{ github.ref_name }} + GOPRIVATE=${{ env.GOPRIVATE }} + secrets: | + github_token=${{ secrets.LBTSM_REPO_TOKEN }} cache-from: type=gha cache-to: type=gha,mode=max diff --git a/Dockerfile b/Dockerfile index fd23864..db1c866 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,8 +6,21 @@ FROM golang:${GO_VERSION}-bookworm AS builder WORKDIR /src +ARG GOPRIVATE=github.com/lbtsm/* +ENV GOPRIVATE=${GOPRIVATE} +ENV GONOSUMDB=${GOPRIVATE} + COPY go.mod go.sum ./ -RUN go mod download +RUN --mount=type=secret,id=github_token \ + set -eu; \ + cleanup() { rm -f /root/.netrc; }; \ + trap cleanup EXIT; \ + if [ -s /run/secrets/github_token ]; then \ + token="$(cat /run/secrets/github_token)"; \ + printf "machine github.com\nlogin x-access-token\npassword %s\n" "$token" > /root/.netrc; \ + chmod 0600 /root/.netrc; \ + fi; \ + go mod download COPY . . diff --git a/README.md b/README.md index 7848be1..a3e4582 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,14 @@ docker compose up -d --build docker compose logs -f bridge-monitor ``` +If the `github.com/lbtsm/*` Go modules are private, set a GitHub token with +read access before building locally: + +```shell +export LBTSM_REPO_TOKEN="github_pat_xxx" +docker compose up -d --build +``` + After pulling updates on the server: ```shell @@ -74,3 +82,7 @@ docker compose up -d --build GitHub Actions builds and tests the project on pull requests, and builds the Docker image on pushes to `main`, `master`, or version tags. Non-PR builds are published to GitHub Container Registry as `ghcr.io//:`. +The default branch also publishes `ghcr.io//:latest`. + +For private `github.com/lbtsm/*` dependencies, add a repository secret named +`LBTSM_REPO_TOKEN` with read access to the private dependency repositories. diff --git a/compose.yaml b/compose.yaml index 04dee8c..a0f19ae 100644 --- a/compose.yaml +++ b/compose.yaml @@ -4,6 +4,8 @@ services: context: . args: VERSION: ${MONITOR_VERSION:-local} + secrets: + - github_token image: ${MONITOR_IMAGE:-bridge-monitor:local} container_name: ${MONITOR_CONTAINER_NAME:-bridge-monitor} restart: unless-stopped @@ -15,3 +17,7 @@ services: volumes: - ${MONITOR_DATA_DIR:-/opt/bridge-monitor}:/app/runtime command: ["monitor", "--config", "/app/runtime/config.json"] + +secrets: + github_token: + environment: LBTSM_REPO_TOKEN From c3a8491736ac779d93b01172d83caac6f9dd045e Mon Sep 17 00:00:00 2001 From: lbtsm Date: Sat, 5 Sep 2026 15:00:40 +0800 Subject: [PATCH 13/13] Add sync height alarm toggle --- README.md | 1 + internal/config/apply.go | 1 + internal/config/apply_test.go | 61 ++++++++++++---------- internal/config/config.go | 88 ++++++++++++++++++-------------- internal/config/config_test.go | 45 ++++++++++++++++ internal/config/const.go | 1 + internal/config/diff_test.go | 26 ++++++++-- internal/config/reloader_test.go | 18 +++++++ pkg/monitor/monitor.go | 13 +++-- pkg/monitor/monitor_test.go | 37 ++++++++++++++ 10 files changed, 216 insertions(+), 75 deletions(-) create mode 100644 internal/config/config_test.go diff --git a/README.md b/README.md index a3e4582..9a4fd72 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ files and keystores out of git. "waterLine": "5000000000000000000", // If the user balance is lower than, an alarm will be triggered, unit : wei "changeInterval": "3000", // How long does the lightnode height remain unchanged, triggering the alarm, use for near unit : seconds "checkHeightCount": "20", // How long does the lightnode height not change remain unchanged, triggering the alarm, default 15 + "syncHeightAlarm": "false", // Optional: disable other-chain-to-map sync height alarm, default true } ``` diff --git a/internal/config/apply.go b/internal/config/apply.go index 75e156d..f0d0bb3 100644 --- a/internal/config/apply.go +++ b/internal/config/apply.go @@ -23,6 +23,7 @@ func ApplyHotReloadable(target, source *OptConfig) { target.WaterLine = source.WaterLine target.LightNode = source.LightNode target.ApiUrl = source.ApiUrl + target.SyncHeightAlarm = source.SyncHeightAlarm target.From = source.From target.Users = source.Users target.ContractToken = source.ContractToken diff --git a/internal/config/apply_test.go b/internal/config/apply_test.go index bf8dd81..31fb109 100644 --- a/internal/config/apply_test.go +++ b/internal/config/apply_test.go @@ -15,20 +15,21 @@ func TestApplyHotReloadable_OverwritesAllReloadableFields(t *testing.T) { oldTss := &Tss{Maintainer: "old-maint"} target := &OptConfig{ - Name: "bsc", - Id: 56, - Endpoint: "http://old", // immutable, must NOT be touched - KeystorePath: "/keys/old", // immutable - WaterLine: "100", - From: []string{"0xold"}, - Users: []From{{Group: "g1", From: "0xa"}}, - ContractToken: []ContractToken{{Address: "0xold-ct"}}, - Energies: []Energy{{Address: "old-en"}}, - Tss: oldTss, - Tk: oldTk, - Genni: oldGenni, - LightNode: common.HexToAddress("0xaaaa"), - ApiUrl: "old-api", + Name: "bsc", + Id: 56, + Endpoint: "http://old", // immutable, must NOT be touched + KeystorePath: "/keys/old", // immutable + WaterLine: "100", + From: []string{"0xold"}, + Users: []From{{Group: "g1", From: "0xa"}}, + ContractToken: []ContractToken{{Address: "0xold-ct"}}, + Energies: []Energy{{Address: "old-en"}}, + Tss: oldTss, + Tk: oldTk, + Genni: oldGenni, + LightNode: common.HexToAddress("0xaaaa"), + ApiUrl: "old-api", + SyncHeightAlarm: true, } newTk := &Token{BridgeAddr: "new-bridge"} @@ -36,20 +37,21 @@ func TestApplyHotReloadable_OverwritesAllReloadableFields(t *testing.T) { newTss := &Tss{Maintainer: "new-maint"} source := &OptConfig{ - Name: "bsc", - Id: 56, - Endpoint: "http://new", // ignored - KeystorePath: "/keys/new", // ignored - WaterLine: "200", - From: []string{"0xnew"}, - Users: []From{{Group: "g2", From: "0xb"}}, - ContractToken: []ContractToken{{Address: "0xnew-ct"}}, - Energies: []Energy{{Address: "new-en"}}, - Tss: newTss, - Tk: newTk, - Genni: newGenni, - LightNode: common.HexToAddress("0xbbbb"), - ApiUrl: "new-api", + Name: "bsc", + Id: 56, + Endpoint: "http://new", // ignored + KeystorePath: "/keys/new", // ignored + WaterLine: "200", + From: []string{"0xnew"}, + Users: []From{{Group: "g2", From: "0xb"}}, + ContractToken: []ContractToken{{Address: "0xnew-ct"}}, + Energies: []Energy{{Address: "new-en"}}, + Tss: newTss, + Tk: newTk, + Genni: newGenni, + LightNode: common.HexToAddress("0xbbbb"), + ApiUrl: "new-api", + SyncHeightAlarm: false, } ApplyHotReloadable(target, source) @@ -84,6 +86,9 @@ func TestApplyHotReloadable_OverwritesAllReloadableFields(t *testing.T) { if target.ApiUrl != "new-api" { t.Errorf("ApiUrl = %q, want new-api", target.ApiUrl) } + if target.SyncHeightAlarm { + t.Error("SyncHeightAlarm = true, want false") + } } // TestApplyHotReloadable_PreservesImmutableFields verifies that fields diff --git a/internal/config/config.go b/internal/config/config.go index 0168d7e..941adbd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -284,50 +284,52 @@ func loadConfig(file string, config *Config) error { } type OptConfig struct { - Name string // Human-readable chain name - Id ChainId // ChainID - Endpoint string // url for rpc endpoint - From []string // address of key to use - KeystorePath string // Location of keyfiles - GasLimit *big.Int - MaxGasPrice *big.Int - GasMultiplier *big.Float - WaterLine string - ChangeInterval string - ApiUrl string - StartBlock *big.Int - MapChainID ChainId - LightNode common.Address // the lightnode to sync header - Tk *Token - Genni *Api - CheckHgtCount int64 - Users []From - ContractToken []ContractToken - Energies []Energy - Tss *Tss + Name string // Human-readable chain name + Id ChainId // ChainID + Endpoint string // url for rpc endpoint + From []string // address of key to use + KeystorePath string // Location of keyfiles + GasLimit *big.Int + MaxGasPrice *big.Int + GasMultiplier *big.Float + WaterLine string + ChangeInterval string + ApiUrl string + StartBlock *big.Int + MapChainID ChainId + LightNode common.Address // the lightnode to sync header + Tk *Token + Genni *Api + CheckHgtCount int64 + SyncHeightAlarm bool + Users []From + ContractToken []ContractToken + Energies []Energy + Tss *Tss } // ParseOptConfig uses a core.ChainConfig to construct a corresponding Config func ParseOptConfig(chainCfg *ChainConfig, tks *Token, genni *Api, users []From) (*OptConfig, error) { config := &OptConfig{ - Id: chainCfg.Id, - From: strings.Split(chainCfg.From, ","), - Name: chainCfg.Name, - Endpoint: chainCfg.Endpoint, - KeystorePath: DefaultKeystorePath, - WaterLine: "", - ChangeInterval: "", - StartBlock: big.NewInt(0), - GasLimit: big.NewInt(DefaultGasLimit), - MaxGasPrice: big.NewInt(DefaultGasPrice), - GasMultiplier: big.NewFloat(DefaultGasMultiplier), - Tk: tks, - Genni: genni, - CheckHgtCount: DefaultCheckHgtCount, - ContractToken: chainCfg.ContractToken, - Energies: chainCfg.Energies, - Users: users, - Tss: chainCfg.Tss, + Id: chainCfg.Id, + From: strings.Split(chainCfg.From, ","), + Name: chainCfg.Name, + Endpoint: chainCfg.Endpoint, + KeystorePath: DefaultKeystorePath, + WaterLine: "", + ChangeInterval: "", + StartBlock: big.NewInt(0), + GasLimit: big.NewInt(DefaultGasLimit), + MaxGasPrice: big.NewInt(DefaultGasPrice), + GasMultiplier: big.NewFloat(DefaultGasMultiplier), + Tk: tks, + Genni: genni, + CheckHgtCount: DefaultCheckHgtCount, + SyncHeightAlarm: true, + ContractToken: chainCfg.ContractToken, + Energies: chainCfg.Energies, + Users: users, + Tss: chainCfg.Tss, } if chainCfg.NearKeystorePath != "" { @@ -367,5 +369,13 @@ func ParseOptConfig(chainCfg *ChainConfig, tks *Token, genni *Api, users []From) config.CheckHgtCount = int64(count) } + if syncHeightAlarm, ok := chainCfg.Opts[SyncHeightAlarm]; ok && syncHeightAlarm != "" { + enabled, err := strconv.ParseBool(syncHeightAlarm) + if err != nil { + return nil, fmt.Errorf("%s must be boolean: %w", SyncHeightAlarm, err) + } + config.SyncHeightAlarm = enabled + } + return config, nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..03de946 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,45 @@ +package config + +import "testing" + +func TestParseOptConfig_SyncHeightAlarmDefaultEnabled(t *testing.T) { + cfg, err := ParseOptConfig(&ChainConfig{ + Name: "klaytn", + Id: 8217, + Endpoint: "http://klaytn.local", + Opts: map[string]string{}, + }, nil, nil, nil) + if err != nil { + t.Fatalf("ParseOptConfig returned error: %v", err) + } + if !cfg.SyncHeightAlarm { + t.Fatal("SyncHeightAlarm = false, want true by default") + } +} + +func TestParseOptConfig_SyncHeightAlarmCanBeDisabled(t *testing.T) { + cfg, err := ParseOptConfig(&ChainConfig{ + Name: "klaytn", + Id: 8217, + Endpoint: "http://klaytn.local", + Opts: map[string]string{SyncHeightAlarm: "false"}, + }, nil, nil, nil) + if err != nil { + t.Fatalf("ParseOptConfig returned error: %v", err) + } + if cfg.SyncHeightAlarm { + t.Fatal("SyncHeightAlarm = true, want false when opts.syncHeightAlarm=false") + } +} + +func TestParseOptConfig_InvalidSyncHeightAlarmRejected(t *testing.T) { + _, err := ParseOptConfig(&ChainConfig{ + Name: "klaytn", + Id: 8217, + Endpoint: "http://klaytn.local", + Opts: map[string]string{SyncHeightAlarm: "nope"}, + }, nil, nil, nil) + if err == nil { + t.Fatal("ParseOptConfig returned nil error, want invalid syncHeightAlarm rejected") + } +} diff --git a/internal/config/const.go b/internal/config/const.go index 875a15b..a59d87b 100644 --- a/internal/config/const.go +++ b/internal/config/const.go @@ -44,6 +44,7 @@ var ( ChangeInterval = "changeInterval" CheckHeightCount = "checkHeightCount" ApiUrl = "apiUrl" + SyncHeightAlarm = "syncHeightAlarm" ) const ( diff --git a/internal/config/diff_test.go b/internal/config/diff_test.go index 770db04..477fe33 100644 --- a/internal/config/diff_test.go +++ b/internal/config/diff_test.go @@ -99,6 +99,24 @@ func TestDiffChains_DataOnlyChangeUpdates(t *testing.T) { } } +func TestDiffChains_SyncHeightAlarmChangeUpdates(t *testing.T) { + old := []RawChainConfig{chainMAP(), chainBSC(func(c *RawChainConfig) { + c.Opts = map[string]string{SyncHeightAlarm: "true"} + })} + new := []RawChainConfig{chainMAP(), chainBSC(func(c *RawChainConfig) { + c.Opts = map[string]string{SyncHeightAlarm: "false"} + })} + + d := DiffChains(old, new) + + if got := names(d.Updates); !reflect.DeepEqual(got, []string{"bsc"}) { + t.Errorf("Updates = %v, want [bsc]", got) + } + if len(d.Restarts) != 0 { + t.Errorf("did not expect Restarts for syncHeightAlarm change") + } +} + func TestDiffChains_NoChangeProducesEmptyDiff(t *testing.T) { chains := []RawChainConfig{chainMAP(), chainBSC()} d := DiffChains(chains, chains) @@ -111,15 +129,15 @@ func TestDiffChains_MixedAddRemoveRestartUpdate(t *testing.T) { old := []RawChainConfig{ chainMAP(), chainBSC(), - {Name: "tron", Id: "728126428", Endpoint: "http://tron.old"}, // restart candidate - {Name: "old-chain", Id: "999", Endpoint: "http://x"}, // remove + {Name: "tron", Id: "728126428", Endpoint: "http://tron.old"}, // restart candidate + {Name: "old-chain", Id: "999", Endpoint: "http://x"}, // remove {Name: "eth", Id: "1", Endpoint: "u", Users: []From{{Group: "g"}}}, // update } new := []RawChainConfig{ chainMAP(), chainBSC(), - {Name: "tron", Id: "728126428", Endpoint: "http://tron.NEW"}, // restart - {Name: "new-chain", Id: "100", Endpoint: "http://y"}, // add + {Name: "tron", Id: "728126428", Endpoint: "http://tron.NEW"}, // restart + {Name: "new-chain", Id: "100", Endpoint: "http://y"}, // add {Name: "eth", Id: "1", Endpoint: "u", Users: []From{{Group: "g2"}}}, // update } diff --git a/internal/config/reloader_test.go b/internal/config/reloader_test.go index 4910978..57ab2c1 100644 --- a/internal/config/reloader_test.go +++ b/internal/config/reloader_test.go @@ -159,6 +159,24 @@ func TestReloadFromFile_RejectsChangeIntervalChange(t *testing.T) { } } +func TestReloadFromFile_AllowsSyncHeightAlarmChange(t *testing.T) { + dir := t.TempDir() + old := validRawConfig() + old.Chains[1].Opts = map[string]string{SyncHeightAlarm: "true"} + store := NewStore(&old) + + updated := validRawConfig() + updated.Chains[1].Opts = map[string]string{SyncHeightAlarm: "false"} + path := writeJSON(t, dir, "config.json", updated) + + if err := ReloadFromFile(store, path); err != nil { + t.Fatalf("syncHeightAlarm change should be allowed, got error: %v", err) + } + if got := store.Load().Chains[1].Opts[SyncHeightAlarm]; got != "false" { + t.Fatalf("syncHeightAlarm = %q, want false", got) + } +} + func TestReloadFromFile_RejectsNameRenameWithoutIdChange(t *testing.T) { // pure rename (chain.id stays same, only name flips) is rejected dir := t.TempDir() diff --git a/pkg/monitor/monitor.go b/pkg/monitor/monitor.go index 5bee631..0086df1 100644 --- a/pkg/monitor/monitor.go +++ b/pkg/monitor/monitor.go @@ -515,20 +515,25 @@ func (m *Monitor) nativeCheck(contract string) { } func (m *Monitor) OtherChainCheck() { - if m.Cfg.LightNode == config.ZeroAddress { + snap := m.Snapshot() + if !snap.SyncHeightAlarm { + m.heightCount = 0 + return + } + if snap.LightNode == config.ZeroAddress { return } - height, err := mapprotocol.Get2MapHeight(m.Cfg.Id) + height, err := mapprotocol.Get2MapHeight(snap.Id) m.Log.Info("Check Height", "syncHeight", height, "record", m.syncedHeight, "heightCount", m.heightCount) if err != nil { m.Log.Error("get2MapHeight failed", "err", err) } else { if m.syncedHeight.Uint64() == height.Uint64() { m.heightCount = m.heightCount + 1 - if m.heightCount >= m.Cfg.CheckHgtCount { + if m.heightCount >= snap.CheckHgtCount { util.Alarm(context.Background(), fmt.Sprintf("Sync Height No change within %d minutes chains=%s, height=%d", - m.Cfg.CheckHgtCount, m.Cfg.Name, height.Uint64())) + snap.CheckHgtCount, snap.Name, height.Uint64())) } } else { m.heightCount = 0 diff --git a/pkg/monitor/monitor_test.go b/pkg/monitor/monitor_test.go index 1736bbe..745a9da 100644 --- a/pkg/monitor/monitor_test.go +++ b/pkg/monitor/monitor_test.go @@ -1,10 +1,14 @@ package monitor import ( + "math/big" "testing" + "github.com/ChainSafe/log15" + "github.com/ethereum/go-ethereum/common" "github.com/mapprotocol/monitor/internal/chain" "github.com/mapprotocol/monitor/internal/config" + "github.com/mapprotocol/monitor/internal/mapprotocol" ) // TestPrepareTick_ReadsLatestWaterLine: each call to prepareTick should @@ -57,3 +61,36 @@ func TestPrepareTick_SnapshotIsIndependent(t *testing.T) { t.Fatalf("snapshot mutated retroactively, WaterLine=%q", snap.WaterLine) } } + +func TestOtherChainCheck_SkipsWhenSyncHeightAlarmDisabled(t *testing.T) { + original := mapprotocol.Get2MapHeight + defer func() { + mapprotocol.Get2MapHeight = original + }() + + called := false + mapprotocol.Get2MapHeight = func(chainID config.ChainId) (*big.Int, error) { + called = true + return big.NewInt(0), nil + } + + cfg := &config.OptConfig{ + Name: "klaytn", + Id: 8217, + LightNode: common.HexToAddress("0x0000000000000000000000000000000000000001"), + CheckHgtCount: 1000, + SyncHeightAlarm: false, + } + m := New(chain.NewCommonSync(nil, cfg, log15.New(), nil, nil)) + m.heightCount = 99 + m.syncedHeight = big.NewInt(123) + + m.OtherChainCheck() + + if called { + t.Fatal("Get2MapHeight was called, want skipped when SyncHeightAlarm=false") + } + if m.heightCount != 0 { + t.Fatalf("heightCount = %d, want reset to 0", m.heightCount) + } +}