From daaed39bac8f67a49bee5d26ba6c6d116e76d8d1 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Thu, 2 Jul 2026 09:57:19 +0800 Subject: [PATCH] =?UTF-8?q?fix(server):=20=E7=94=A8=E9=87=8F=E5=88=86?= =?UTF-8?q?=E9=92=9F=E6=95=B0=E6=8C=89=E5=A2=99=E4=B8=8A=E6=97=B6=E9=92=9F?= =?UTF-8?q?=E5=8E=BB=E9=87=8D=20=E2=80=94=E2=80=94=20=E5=A4=9A=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E5=B9=B6=E5=8F=91=E4=B8=8D=E5=86=8D=20N=C3=97=20?= =?UTF-8?q?=E8=B6=85=E8=AE=A1(bug)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent 按每 dp_uuid(每设备)每窗口计 SessionMinutes=1,控制面 ReportUsage 逐条累加进 usage_daily.minutes_used → N 台并发 = N×墙上时钟(线上 chenxin 2903 分钟/天,不可能)。 免费额度是「所有设备共同的时间」,超计会把免费账号错误判「时长耗尽」而断连。 改 ReportUsage:账户维度按 user 分组去重 —— 字节仍求和(可加),分钟按窗口取 max (=窗口墙上分钟,恒为 1)去重,循环后每 user 各 flush 一次 AccumulateUsage/Hourly。 每设备维度(usage_device_*)保持逐台累加不变。usage_daily.minutes_used ≤ Σ 每设备分钟, /me 的 quota_today_min 与 ConnectNode 免费门随之正确。历史数据次日归零、不回填。 Co-Authored-By: Claude Opus 4.8 --- server/internal/nodes/grpc_test.go | 66 +++++++++++++++++++++++++++ server/internal/nodes/handler_grpc.go | 41 +++++++++++------ 2 files changed, 94 insertions(+), 13 deletions(-) diff --git a/server/internal/nodes/grpc_test.go b/server/internal/nodes/grpc_test.go index c174ce4..f165cf3 100644 --- a/server/internal/nodes/grpc_test.go +++ b/server/internal/nodes/grpc_test.go @@ -942,6 +942,72 @@ func TestReportUsage_Accumulates(t *testing.T) { } } +// TestReportUsage_AccountMinutesDedupPerUser verifies the account-level minutes are +// deduped to WALL-CLOCK per user within a window: two devices of the same user active +// in one window count as 1 minute (not 2), while bytes still sum and each device keeps +// its own minute. Guards the多设备并发超计 fix. +func TestReportUsage_AccountMinutesDedupPerUser(t *testing.T) { + const nodeUUID = "test-node-usage-dedup" + b := newTestServer(t, 1, nodeUUID) + ctx := context.Background() + // Two devices of user 101 (dev 55 & 66) + one device of user 202 (dev 77). + b.store.devicesByDpUUID = map[string][2]int64{ + "dp-u1-d1": {101, 55}, + "dp-u1-d2": {101, 66}, + "dp-u2-d1": {202, 77}, + } + + _, _, conn := enrollNode(t, b, nodeUUID) + client := agentv1.NewAgentServiceClient(conn) + if _, err := client.Register(ctx, &agentv1.RegisterRequest{NodeUUID: nodeUUID}); err != nil { + t.Fatalf("Register: %v", err) + } + + now := time.Now() + if _, err := client.ReportUsage(ctx, &agentv1.UsageReport{ + NodeUUID: nodeUUID, + WindowStartUnix: now.Add(-time.Minute).Unix(), + WindowEndUnix: now.Unix(), + Entries: []*agentv1.UsageEntry{ + {DpUUID: "dp-u1-d1", BytesUp: 100, BytesDown: 200, SessionMinutes: 1}, + {DpUUID: "dp-u1-d2", BytesUp: 10, BytesDown: 20, SessionMinutes: 1}, + {DpUUID: "dp-u2-d1", BytesUp: 5, BytesDown: 7, SessionMinutes: 1}, + }, + }); err != nil { + t.Fatalf("ReportUsage: %v", err) + } + + // Account rollup: one entry per user; user 101 minutes deduped to 1 (not 2), + // bytes summed across its two devices; user 202 = 1 minute. + byUser := map[int64]mockUsageEntry{} + for _, e := range b.store.usageLog() { + byUser[e.UserID] = e + } + if len(byUser) != 2 { + t.Fatalf("account rollup should have 2 users, got %d", len(byUser)) + } + if u := byUser[101]; u.Minutes != 1 || u.BytesUp != 110 || u.BytesDown != 220 { + t.Errorf("user101 account: minutes=%d bytesUp=%d bytesDown=%d, want 1/110/220 (墙上时钟去重)", + u.Minutes, u.BytesUp, u.BytesDown) + } + if u := byUser[202]; u.Minutes != 1 || u.BytesUp != 5 { + t.Errorf("user202 account: minutes=%d bytesUp=%d, want 1/5", u.Minutes, u.BytesUp) + } + + // Per-device: each of user101's two devices keeps its own minute (2 rows, 2 min total). + var u1Rows int + var u1Min int64 + for _, d := range b.store.deviceUsageLog() { + if d.UserID == 101 { + u1Rows++ + u1Min += d.Minutes + } + } + if u1Rows != 2 || u1Min != 2 { + t.Errorf("user101 per-device: rows=%d totalMinutes=%d, want 2/2 (每设备各计 1)", u1Rows, u1Min) + } +} + // TestReportUsage_PerDevice verifies a per-device dp_uuid is dual-written: account // rollup (usage_daily) AND per-device attribution (usage_device_daily). func TestReportUsage_PerDevice(t *testing.T) { diff --git a/server/internal/nodes/handler_grpc.go b/server/internal/nodes/handler_grpc.go index 0eb80fe..c7f630b 100644 --- a/server/internal/nodes/handler_grpc.go +++ b/server/internal/nodes/handler_grpc.go @@ -290,6 +290,13 @@ func (h *Handler) ReportUsage(ctx context.Context, req *agentv1.UsageReport) (*a windowEnd := time.Unix(req.WindowEndUnix, 0).UTC() date := windowEnd.Truncate(24 * time.Hour) + // Account-level minutes are deduped to WALL-CLOCK per user: the免费额度是「所有 + // 设备共同的时间」,同一账户多台设备在同一窗口都活跃时只能算 1 分钟,否则 N 台并发 → + // minutes_used = N×墙上时钟(超计,会把免费账号错误判耗尽)。字节是可加的 → 求和; + // 分钟按窗口取 max(= 窗口墙上分钟)去重。每设备维度(usage_device_*)仍逐台累加。 + type acctAgg struct{ bytesUp, bytesDown, minutes int64 } + byUser := make(map[int64]*acctAgg) + for _, entry := range req.Entries { if entry.DpUUID == "" { continue @@ -303,21 +310,18 @@ func (h *Handler) ReportUsage(ctx context.Context, req *agentv1.UsageReport) (*a if !found { continue } - // Account-level rollup (always): daily (quota/today) + hourly (tz-aware - // display curve, keyed by the window-end's UTC hour). - if err := h.store.AccumulateUsage(ctx, userID, date, - entry.BytesUp, entry.BytesDown, entry.SessionMinutes, - ); err != nil { - slog.Warn("nodes.Handler.ReportUsage: accumulate failed", - "user_id", userID, "err", err) + // Fold into the per-user account aggregate (bytes sum, minutes max = 墙上时钟去重). + a := byUser[userID] + if a == nil { + a = &acctAgg{} + byUser[userID] = a } - if err := h.store.AccumulateHourly(ctx, userID, windowEnd, - entry.BytesUp, entry.BytesDown, entry.SessionMinutes, - ); err != nil { - slog.Warn("nodes.Handler.ReportUsage: hourly accumulate failed", - "user_id", userID, "err", err) + a.bytesUp += entry.BytesUp + a.bytesDown += entry.BytesDown + if entry.SessionMinutes > a.minutes { + a.minutes = entry.SessionMinutes } - // Per-device attribution (only when the dp_uuid maps to a registered device; + // Per-device attribution (per entry — each device's own bytes/minutes; // deviceID==0 means a legacy account-level credential — no device dimension). if deviceID > 0 { if err := h.store.AccumulateDeviceUsage(ctx, userID, deviceID, date, @@ -340,5 +344,16 @@ func (h *Handler) ReportUsage(ctx context.Context, req *agentv1.UsageReport) (*a } } } + + // Flush the account-level rollup once per user (deduped wall-clock minutes): + // daily (quota/today) + hourly (tz-aware display curve, keyed by window-end's UTC hour). + for userID, a := range byUser { + if err := h.store.AccumulateUsage(ctx, userID, date, a.bytesUp, a.bytesDown, a.minutes); err != nil { + slog.Warn("nodes.Handler.ReportUsage: accumulate failed", "user_id", userID, "err", err) + } + if err := h.store.AccumulateHourly(ctx, userID, windowEnd, a.bytesUp, a.bytesDown, a.minutes); err != nil { + slog.Warn("nodes.Handler.ReportUsage: hourly accumulate failed", "user_id", userID, "err", err) + } + } return &agentv1.UsageAck{}, nil }