diff --git a/cmd/dashboard/controller/controller.go b/cmd/dashboard/controller/controller.go index a3dd50b0..1668e192 100644 --- a/cmd/dashboard/controller/controller.go +++ b/cmd/dashboard/controller/controller.go @@ -100,6 +100,7 @@ func routers(r *gin.Engine, frontendDist fs.FS) { optionalAuth.GET("/service/:id/history", restScopeMiddleware(model.ScopeServiceRead), commonHandler(getServiceHistory)) optionalAuth.GET("/server/:id/service", restScopeMiddleware(model.ScopeServiceRead), commonHandler(listServerServices)) optionalAuth.GET("/server/:id/metrics", restScopeMiddleware(model.ScopeServerRead), commonHandler(getServerMetrics)) + optionalAuth.GET("/server/:id/online-history", restScopeMiddleware(model.ScopeServerRead), commonHandler(getServerOnlineHistory)) // CSRF middleware applies group-wide. Safe methods short-circuit and // PAT bearer requests bypass — so the only callers gated are diff --git a/cmd/dashboard/controller/online_history.go b/cmd/dashboard/controller/online_history.go new file mode 100644 index 00000000..302f1029 --- /dev/null +++ b/cmd/dashboard/controller/online_history.go @@ -0,0 +1,121 @@ +package controller + +import ( + "strconv" + "time" + + "github.com/gin-gonic/gin" + + "github.com/nezhahq/nezha/model" + "github.com/nezhahq/nezha/service/singleton" +) + +// Get server online history +// @Summary Get server online history +// @Security BearerAuth +// @Schemes +// @Description Get online periods and aggregated uptime statistics of a specific server +// @Tags common +// @Param id path uint true "Server ID" +// @Param days query int false "Query window in days (default: configured online_stats_days, clamped to it)" +// @Produce json +// @Success 200 {object} model.CommonResponse[model.OnlineHistoryResponse] +// @Router /server/{id}/online-history [get] +func getServerOnlineHistory(c *gin.Context) (*model.OnlineHistoryResponse, error) { + idStr := c.Param("id") + serverID, err := strconv.ParseUint(idStr, 10, 64) + if err != nil { + return nil, err + } + + server, ok := singleton.ServerShared.Get(serverID) + if !ok || server == nil { + return nil, singleton.Localizer.ErrorT("server not found") + } + if !userCanViewServer(c, server) { + return nil, singleton.Localizer.ErrorT("unauthorized") + } + _, isMember := c.Get(model.CtxKeyAuthorizedUser) + + maxDays := singleton.Conf.OnlineHistoryRetentionDays() + if maxDays < model.MinOnlineStatsDays || maxDays > model.MaxOnlineStatsDays { + maxDays = model.DefaultOnlineStatsDays + } + days := maxDays + if daysStr := c.Query("days"); daysStr != "" { + d, err := strconv.Atoi(daysStr) + if err != nil || d < 1 { + return nil, singleton.Localizer.ErrorT("invalid days parameter") + } + // 查询窗口收敛到后台设置的统计范围内 + if d > maxDays { + d = maxDays + } + days = d + } + + if !isMember && days > 1 { + return nil, singleton.Localizer.ErrorT("unauthorized: only 1d data available for guests") + } + + now := time.Now() + windowEnd := now + windowStart := model.OnlineHistoryWindowStart(now, days, singleton.Loc) + + sessions, err := singleton.OnlineHistoryShared.Sessions(serverID, windowStart, windowEnd) + if err != nil { + return nil, newGormError("%v", err) + } + + periods := model.ClipOnlineSessions(sessions, windowStart, windowEnd) + daily := model.DailyOnlineBreakdown(periods, windowStart, windowEnd, singleton.Loc) + // 总时长取各日之和，与 daily[] 严格对账（每日累计原始时长后统一取整， + // 口径见 DailyOnlineBreakdown） + var totalSeconds uint64 + for _, d := range daily { + totalSeconds += d.OnlineSeconds + } + windowSeconds := uint64(windowEnd.Sub(windowStart).Seconds()) + onlineRate := 0.0 + if windowSeconds > 0 { + onlineRate = float64(totalSeconds) / float64(windowSeconds) * 100 + } + + runtime := server.RuntimeSnapshot() + currentlyOnline := !runtime.LastActive.IsZero() && now.Sub(runtime.LastActive) <= model.OnlineActiveWindow + var lastActive *time.Time + if !runtime.LastActive.IsZero() { + lastActiveUTC := runtime.LastActive.UTC() + lastActive = &lastActiveUTC + } + + // 响应统一以 UTC 序列化，避免窗口时间与库中读回的会话时间混用 + // 本地偏移与 Z 两种写法（时段为值拷贝，不影响调用方） + utcPeriods := make([]model.OnlinePeriod, len(periods)) + for i, p := range periods { + utcPeriods[i] = model.OnlinePeriod{Start: p.Start.UTC(), End: p.End.UTC(), Ongoing: p.Ongoing} + } + + return &model.OnlineHistoryResponse{ + ServerID: serverID, + ServerName: server.Name, + Days: days, + MaxDays: maxDays, + Timezone: singleton.Loc.String(), + TzOffset: tzOffsetSeconds(now, singleton.Loc), + WindowStart: windowStart.UTC(), + WindowEnd: windowEnd.UTC(), + TotalOnlineSeconds: totalSeconds, + OnlineRate: onlineRate, + CurrentlyOnline: currentlyOnline, + LastActive: lastActive, + Periods: utcPeriods, + Daily: daily, + }, nil +} + +// tzOffsetSeconds 返回指定时区在给定时刻相对 UTC 的偏移秒数。 +func tzOffsetSeconds(at time.Time, loc *time.Location) int { + _, offset := at.In(loc).Zone() + return offset +} diff --git a/cmd/dashboard/controller/online_history_test.go b/cmd/dashboard/controller/online_history_test.go new file mode 100644 index 00000000..544703a4 --- /dev/null +++ b/cmd/dashboard/controller/online_history_test.go @@ -0,0 +1,356 @@ +package controller + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + + "github.com/nezhahq/nezha/model" + "github.com/nezhahq/nezha/pkg/i18n" + "github.com/nezhahq/nezha/service/singleton" +) + +func setupOnlineHistoryControllerTest(t *testing.T) (visibleServer, hiddenServer *model.Server) { + t.Helper() + originalDB := singleton.DB + originalLocalizer := singleton.Localizer + originalServer := singleton.ServerShared + originalConf := singleton.Conf + originalLoc := singleton.Loc + if singleton.Localizer == nil { + singleton.Localizer = i18n.NewLocalizer("en_US", "nezha", "translations", i18n.Translations) + } + + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + require.NoError(t, err) + sqlDB, err := db.DB() + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.User{}, &model.Server{}, &model.OnlineSession{})) + + singleton.DB = db + singleton.ServerShared = singleton.NewEmptyServerClassForTest() + singleton.Conf = &singleton.ConfigClass{Config: &model.Config{}} + singleton.Conf.OnlineStatsDays = 30 + singleton.Loc = time.FixedZone("TEST", 8*3600) + + visibleServer = &model.Server{ + Common: model.Common{ID: 1}, + Name: "visible", + UUID: "online-history-visible", + LastActive: time.Now(), + } + hiddenServer = &model.Server{ + Common: model.Common{ID: 2}, + Name: "hidden", + UUID: "online-history-hidden", + HideForGuest: true, + } + singleton.ServerShared.InsertForTest(visibleServer) + singleton.ServerShared.InsertForTest(hiddenServer) + + t.Cleanup(func() { + _ = sqlDB.Close() + singleton.DB = originalDB + singleton.Localizer = originalLocalizer + singleton.ServerShared = originalServer + singleton.Conf = originalConf + singleton.Loc = originalLoc + }) + return visibleServer, hiddenServer +} + +func onlineHistoryCtx(t *testing.T, serverID string, query string) *gin.Context { + t.Helper() + gin.SetMode(gin.TestMode) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodGet, "/api/v1/server/"+serverID+"/online-history?"+query, nil) + c.Params = gin.Params{{Key: "id", Value: serverID}} + return c +} + +func onlineHistoryCtxAsMember(t *testing.T, serverID string, query string) *gin.Context { + t.Helper() + c := onlineHistoryCtx(t, serverID, query) + c.Set(model.CtxKeyAuthorizedUser, &model.User{Common: model.Common{ID: 100}, Role: model.RoleMember}) + return c +} + +func seedOnlineSessions(t *testing.T, serverID uint64, now time.Time) []model.OnlineSession { + t.Helper() + now = now.Truncate(time.Second) + sessions := []model.OnlineSession{ + {ServerID: serverID, StartedAt: now.Add(-48 * time.Hour), EndedAt: now.Add(-47 * time.Hour), LastActiveAt: now.Add(-47 * time.Hour)}, + {ServerID: serverID, StartedAt: now.Add(-25 * time.Hour), EndedAt: now.Add(-23*time.Hour - 30*time.Minute), LastActiveAt: now.Add(-23 * time.Hour)}, + {ServerID: serverID, StartedAt: now.Add(-10 * time.Minute), Ongoing: true, LastActiveAt: now}, + // 超出默认 7 天窗口、但在 30 天窗口内的会话 + {ServerID: serverID, StartedAt: now.Add(-20 * time.Hour * 24), EndedAt: now.Add(-20*time.Hour*24 + time.Hour), LastActiveAt: now.Add(-20 * time.Hour * 24)}, + } + for i := range sessions { + require.NoError(t, singleton.DB.Create(&sessions[i]).Error) + } + return sessions +} + +func TestGetServerOnlineHistoryDefaultsToConfiguredWindow(t *testing.T) { + visible, _ := setupOnlineHistoryControllerTest(t) + now := time.Now() + seedOnlineSessions(t, visible.ID, now) + + // 不带 days 参数：使用后台设置的统计范围（30 天） + c := onlineHistoryCtxAsMember(t, "1", "") + resp, err := getServerOnlineHistory(c) + require.NoError(t, err) + require.Equal(t, 30, resp.Days) + require.Equal(t, 30, resp.MaxDays) + // 4 个会话全部落在 30 天窗口内 + require.Len(t, resp.Periods, 4) + require.True(t, resp.CurrentlyOnline) +} + +func TestGetServerOnlineHistoryDefaultIsSevenDays(t *testing.T) { + visible, _ := setupOnlineHistoryControllerTest(t) + now := time.Now() + seedOnlineSessions(t, visible.ID, now) + + // 出厂默认统计范围为 7 天 + singleton.Conf.OnlineStatsDays = model.DefaultOnlineStatsDays + c := onlineHistoryCtxAsMember(t, "1", "") + resp, err := getServerOnlineHistory(c) + require.NoError(t, err) + require.Equal(t, 7, resp.Days) + // 20 天前的会话不进入 7 天窗口 + require.Len(t, resp.Periods, 3) +} + +func TestGetServerOnlineHistoryAggregatesAndClampsWindow(t *testing.T) { + visible, _ := setupOnlineHistoryControllerTest(t) + now := time.Now() + seedOnlineSessions(t, visible.ID, now) + + c := onlineHistoryCtxAsMember(t, "1", "days=9999") + resp, err := getServerOnlineHistory(c) + require.NoError(t, err) + // 请求超出后台设置窗口时收敛到设置值 + require.Equal(t, 30, resp.Days) + + // 聚合正确性：三个已结束/进行中会话都计入，进行中会话按当前时间截断 + var total uint64 + for _, p := range resp.Periods { + total += uint64(p.End.Sub(p.Start).Seconds()) + } + require.Equal(t, resp.TotalOnlineSeconds, total) + require.GreaterOrEqual(t, resp.TotalOnlineSeconds, uint64(3600+5400+599)) + + // 每日拆分合计等于总时长 + var dailySum uint64 + for _, d := range resp.Daily { + dailySum += d.OnlineSeconds + } + require.Equal(t, resp.TotalOnlineSeconds, dailySum) + + // 在线率按窗口秒数计算 + windowSeconds := resp.WindowEnd.Sub(resp.WindowStart).Seconds() + require.InDelta(t, float64(resp.TotalOnlineSeconds)/windowSeconds*100, resp.OnlineRate, 0.001) +} + +func TestGetServerOnlineHistoryInvalidDays(t *testing.T) { + setupOnlineHistoryControllerTest(t) + for _, q := range []string{"days=abc", "days=0", "days=-1"} { + c := onlineHistoryCtxAsMember(t, "1", q) + _, err := getServerOnlineHistory(c) + require.EqualError(t, err, singleton.Localizer.ErrorT("invalid days parameter").Error(), "query %q should be rejected", q) + } +} + +func TestGetServerOnlineHistoryUnknownServer(t *testing.T) { + setupOnlineHistoryControllerTest(t) + c := onlineHistoryCtx(t, "424242", "") + _, err := getServerOnlineHistory(c) + require.Error(t, err) +} + +func TestGetServerOnlineHistoryGuestLimitedToOneDay(t *testing.T) { + visible, _ := setupOnlineHistoryControllerTest(t) + now := time.Now() + seedOnlineSessions(t, visible.ID, now) + + // 游客不带 days 参数时默认为配置窗口（>1 天），与 metrics 的游客语义一致被拒绝 + _, err := getServerOnlineHistory(onlineHistoryCtx(t, "1", "")) + require.Error(t, err) + + // 游客显式请求多天被拒绝，1 天可用 + _, err = getServerOnlineHistory(onlineHistoryCtx(t, "1", "days=3")) + require.Error(t, err) + resp, err := getServerOnlineHistory(onlineHistoryCtx(t, "1", "days=1")) + require.NoError(t, err) + require.Equal(t, 1, resp.Days) + // 与 1 天窗口相交的两个会话：一个被裁剪到窗口起点，一个进行中 + require.Len(t, resp.Periods, 2) + + // 登录用户不受游客窗口限制 + resp, err = getServerOnlineHistory(onlineHistoryCtxAsMember(t, "1", "days=30")) + require.NoError(t, err) + require.Equal(t, 30, resp.Days) + require.Len(t, resp.Periods, 4) +} + +func TestGetServerOnlineHistoryPermission(t *testing.T) { + visible, _ := setupOnlineHistoryControllerTest(t) + + // 游客可以查看未隐藏服务器（窗口受限于 1 天） + guest := onlineHistoryCtx(t, "1", "days=1") + resp, err := getServerOnlineHistory(guest) + require.NoError(t, err) + require.Equal(t, visible.ID, resp.ServerID) + + // 游客不能查看 HideForGuest 服务器 + guestHidden := onlineHistoryCtx(t, "2", "days=1") + _, err = getServerOnlineHistory(guestHidden) + require.EqualError(t, err, singleton.Localizer.ErrorT("unauthorized").Error()) + + // 成员不能查看他人隐藏服务器 + member := onlineHistoryCtx(t, "2", "days=7") + member.Set(model.CtxKeyAuthorizedUser, &model.User{Common: model.Common{ID: 100}, Role: model.RoleMember}) + _, err = getServerOnlineHistory(member) + require.Error(t, err) +} + +func patchOnlineStatsDays(t *testing.T, days any) error { + t.Helper() + gin.SetMode(gin.TestMode) + body, err := json.Marshal(map[string]any{"user_template": "user-dist", "online_stats_days": days}) + require.NoError(t, err) + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest(http.MethodPatch, "/api/v1/setting", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + _, err = updateConfig(c) + return err +} + +func TestUpdateConfigOnlineStatsDaysRange(t *testing.T) { + originalLocalizer := singleton.Localizer + if singleton.Localizer == nil { + singleton.Localizer = i18n.NewLocalizer("en_US", "nezha", "translations", i18n.Translations) + } + t.Cleanup(func() { singleton.Localizer = originalLocalizer }) + + templates := []model.FrontendTemplate{{Path: "user-dist", IsAdmin: false}} + originalFrontendTemplates := singleton.FrontendTemplates + singleton.FrontendTemplates = templates + t.Cleanup(func() { singleton.FrontendTemplates = originalFrontendTemplates }) + conf := &singleton.ConfigClass{Config: &model.Config{}} + configPath := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, conf.Read(configPath, templates)) + originalConf := singleton.Conf + singleton.Conf = conf + t.Cleanup(func() { singleton.Conf = originalConf }) + + require.Equal(t, model.DefaultOnlineStatsDays, conf.OnlineStatsDays) + + // 合法边界值 + for _, d := range []int{model.MinOnlineStatsDays, 30, model.MaxOnlineStatsDays} { + require.NoError(t, patchOnlineStatsDays(t, d), "days=%d should be accepted", d) + require.Equal(t, d, conf.OnlineStatsDays) + } + + // 越界值整体拒绝且不修改现值 + for _, d := range []int{6, model.MaxOnlineStatsDays + 1, 0, -5} { + require.Error(t, patchOnlineStatsDays(t, d), "days=%d should be rejected", d) + require.Equal(t, model.MaxOnlineStatsDays, conf.OnlineStatsDays) + } + + // 非法值必须整体拒绝：校验先于任何对 Conf 的写入完成，其他字段不得 + // 半提交（评审复现场景：错误返回后 SiteName 不能被请求内容覆盖） + conf.SiteName = "existing-site" + require.Error(t, patchOnlineStatsDays(t, 6)) + require.Equal(t, "existing-site", conf.SiteName) + conf.SiteName = "" + + // 字段缺省时保留现值（部分 PATCH 语义） + require.NoError(t, patchOnlineStatsDays(t, nil)) + require.Equal(t, model.MaxOnlineStatsDays, conf.OnlineStatsDays) + + // 重启持久化断言：updateConfig 显式 Save 后，重新从磁盘读取的配置 + // 必须带上刚写入的值（持久化不再依赖 MCP 过渡块的副作用） + reloaded := &singleton.ConfigClass{Config: &model.Config{}} + require.NoError(t, reloaded.Read(configPath, templates)) + require.Equal(t, model.MaxOnlineStatsDays, reloaded.OnlineStatsDays) +} + +func TestSettingFormOnlineStatsDaysDecode(t *testing.T) { + var absent model.SettingForm + require.NoError(t, json.NewDecoder(bytes.NewReader([]byte(`{"site_name":"x"}`))).Decode(&absent)) + require.Nil(t, absent.OnlineStatsDays, "omitted key must decode to nil") + + var explicit model.SettingForm + require.NoError(t, json.NewDecoder(bytes.NewReader([]byte(`{"online_stats_days":3650}`))).Decode(&explicit)) + require.NotNil(t, explicit.OnlineStatsDays) + require.Equal(t, 3650, *explicit.OnlineStatsDays) +} + +func TestOnlineHistoryReadsPendingCloseBoundary(t *testing.T) { + setupOnlineHistoryControllerTest(t) + previous := singleton.OnlineHistoryShared + t.Cleanup(func() { singleton.OnlineHistoryShared = previous }) + r := singleton.NewOnlineHistoryRecorder() + singleton.OnlineHistoryShared = r + start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) + r.RecordOnline(1, start) + require.NoError(t, singleton.DB.Exec(`CREATE TRIGGER reject_close BEFORE UPDATE ON online_sessions WHEN NEW.ongoing = 0 BEGIN SELECT RAISE(ABORT, 'close failure'); END`).Error) + r.RecordOffline(1, start.Add(time.Minute)) + r.RecordOnline(1, start.Add(30*time.Minute)) + resp, err := getServerOnlineHistory(onlineHistoryCtx(t, "1", "days=1")) + require.NoError(t, err) + require.Len(t, resp.Periods, 2) + require.False(t, resp.Periods[0].Ongoing) + require.True(t, resp.Periods[0].End.Equal(start.Add(time.Minute))) + require.Less(t, resp.TotalOnlineSeconds, uint64(1900)) +} + +func TestOnlineHistoryQueryWithNonUTCSession(t *testing.T) { + setupOnlineHistoryControllerTest(t) + now := time.Now() + offset := time.FixedZone("UTC+12", 12*3600) + session := model.OnlineSession{ServerID: 1, StartedAt: now.Add(-time.Hour).In(offset), LastActiveAt: now.Add(-time.Minute).In(offset), Ongoing: true} + require.NoError(t, singleton.DB.Create(&session).Error) + resp, err := getServerOnlineHistory(onlineHistoryCtx(t, "1", "days=1")) + require.NoError(t, err) + require.Len(t, resp.Periods, 1) + require.True(t, resp.Periods[0].Start.Equal(session.StartedAt)) + require.InDelta(t, 3600, resp.TotalOnlineSeconds, 2) +} + +func TestUpdateConfigFailedSavePreservesRetention(t *testing.T) { + server, _ := setupOnlineHistoryControllerTest(t) + require.NoError(t, singleton.DB.Create(server).Error) + templates := []model.FrontendTemplate{{Path: "user-dist"}} + previous := singleton.FrontendTemplates + singleton.FrontendTemplates = templates + t.Cleanup(func() { singleton.FrontendTemplates = previous }) + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, singleton.Conf.Read(path, templates)) + singleton.Conf.OnlineStatsDays = 30 + require.NoError(t, singleton.Conf.Save()) + // Replacing the destination with a directory deterministically fails Save even as root. + require.NoError(t, os.Remove(path)) + require.NoError(t, os.Mkdir(path, 0700)) + now := time.Now().UTC() + old := model.OnlineSession{ServerID: 1, StartedAt: now.AddDate(0, 0, -15), EndedAt: now.AddDate(0, 0, -14)} + require.NoError(t, singleton.DB.Create(&old).Error) + require.Error(t, patchOnlineStatsDays(t, 7)) + require.Equal(t, 30, singleton.Conf.OnlineStatsDays) + singleton.PruneOnlineSessions() + var retained model.OnlineSession + require.NoError(t, singleton.DB.First(&retained, old.ID).Error) + require.Equal(t, old.ID, retained.ID) +} diff --git a/cmd/dashboard/controller/setting.go b/cmd/dashboard/controller/setting.go index 0b6b70e6..4b5395b0 100644 --- a/cmd/dashboard/controller/setting.go +++ b/cmd/dashboard/controller/setting.go @@ -2,6 +2,7 @@ package controller import ( "errors" + "fmt" "log" "strings" @@ -29,7 +30,7 @@ func listConfig(c *gin.Context) (*model.SettingResponse, error) { isAdmin = user.Role.IsAdmin() } - config := *singleton.Conf + config := singleton.Conf.SettingsSnapshot() config.Language = strings.ReplaceAll(config.Language, "_", "-") conf := model.SettingResponse{ @@ -92,6 +93,12 @@ func updateConfig(c *gin.Context) (any, error) { if !userTemplateValid { return nil, errors.New("invalid user template") } + // 全部输入校验必须先于任何对 singleton.Conf 的写入：任一字段非法都应 + // 整体拒绝，避免"请求以错误收场、先处理的字段却已生效"的半提交状态 + if sf.OnlineStatsDays != nil && + (*sf.OnlineStatsDays < model.MinOnlineStatsDays || *sf.OnlineStatsDays > model.MaxOnlineStatsDays) { + return nil, fmt.Errorf("online_stats_days must be between %d and %d", model.MinOnlineStatsDays, model.MaxOnlineStatsDays) + } singleton.Conf.Language = strings.ReplaceAll(sf.Language, "-", "_") @@ -114,18 +121,23 @@ func updateConfig(c *gin.Context) (any, error) { singleton.Conf.AgentRealIPHeader = sf.AgentRealIPHeader singleton.Conf.AgentTLS = sf.AgentTLS singleton.Conf.UserTemplate = sf.UserTemplate - mcpWasEnabled := singleton.Conf.MCPEnabled() - mcpNext := resolveSettingEnableMCP(sf.EnableMCP, mcpWasEnabled) - if err := applyEnableMCPTransition( - mcpWasEnabled, mcpNext, - singleton.Conf.SetMCPEnabled, - singleton.Conf.Save, - fireMCPKillSwitch, - ); err != nil { + if err := singleton.Conf.CommitOnlineHistoryRetention(sf.OnlineStatsDays, func() error { + mcpWasEnabled := singleton.Conf.MCPEnabled() + mcpNext := resolveSettingEnableMCP(sf.EnableMCP, mcpWasEnabled) + return applyEnableMCPTransition( + mcpWasEnabled, mcpNext, + singleton.Conf.SetMCPEnabled, + singleton.Conf.Save, + fireMCPKillSwitch, + ) + }); err != nil { return nil, newGormError("%v", err) } + // applyEnableMCPTransition already persisted every setting, even when MCP + // is unchanged. A second save could fail after irreversible cleanup. + singleton.OnUpdateLang(singleton.Conf.Language) return nil, nil } diff --git a/cmd/dashboard/controller/setting_single_commit_test.go b/cmd/dashboard/controller/setting_single_commit_test.go new file mode 100644 index 00000000..ee200526 --- /dev/null +++ b/cmd/dashboard/controller/setting_single_commit_test.go @@ -0,0 +1,49 @@ +package controller + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" + + "github.com/stretchr/testify/require" +) + +// The transition helper owns Save (including unchanged MCP settings). A direct +// second Save after it can fail after persistence and irreversible revocation. +func TestUpdateConfigHasSinglePersistenceOwner(t *testing.T) { + file, err := parser.ParseFile(token.NewFileSet(), "setting.go", nil, 0) + require.NoError(t, err) + transitions, directSaves := 0, 0 + found := false + for _, decl := range file.Decls { + fn, ok := decl.(*ast.FuncDecl) + if !ok || fn.Name.Name != "updateConfig" { + continue + } + found = true + ast.Inspect(fn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + if ident, ok := call.Fun.(*ast.Ident); ok && ident.Name == "applyEnableMCPTransition" { + transitions++ + } + if selector, ok := call.Fun.(*ast.SelectorExpr); ok && selector.Sel.Name == "Save" { + directSaves++ + } + return true + }) + } + require.True(t, found) + require.Equal(t, 1, transitions) + require.Zero(t, directSaves) +} + +func TestUnchangedMCPTransitionStillSavesOnce(t *testing.T) { + saves := 0 + err := applyEnableMCPTransition(true, true, func(bool) {}, func() error { saves++; return nil }, func() { t.Error("unchanged MCP must not revoke streams") }) + require.NoError(t, err) + require.Equal(t, 1, saves) +} diff --git a/cmd/dashboard/http_header_timeout.go b/cmd/dashboard/http_header_timeout.go new file mode 100644 index 00000000..3db55188 --- /dev/null +++ b/cmd/dashboard/http_header_timeout.go @@ -0,0 +1,14 @@ +package main + +import ( + "net/http" + "time" +) + +const dashboardReadHeaderTimeout = 5 * time.Second + +func newDashboardHTTPServer(handler http.Handler) *http.Server { + // ReadHeaderTimeout protects incomplete HTTP/1 requests and is cleared after + // headers. Unlike ReadTimeout it does not impose a lifetime on HTTP/2 streams. + return &http.Server{Handler: handler, ReadHeaderTimeout: dashboardReadHeaderTimeout} +} diff --git a/cmd/dashboard/http_header_timeout_test.go b/cmd/dashboard/http_header_timeout_test.go new file mode 100644 index 00000000..d643afaa --- /dev/null +++ b/cmd/dashboard/http_header_timeout_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "crypto/tls" + "io" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDashboardRejectsIncompleteHTTPHeaders(t *testing.T) { + for _, useTLS := range []bool{false, true} { + name := "http" + if useTLS { + name = "https" + } + t.Run(name, func(t *testing.T) { + server := httptest.NewUnstartedServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { t.Error("incomplete headers reached handler") })) + server.Config = newDashboardHTTPServer(server.Config.Handler) + require.Positive(t, server.Config.ReadHeaderTimeout) + server.Config.ReadHeaderTimeout = 50 * time.Millisecond + if useTLS { + server.StartTLS() + } else { + server.Start() + } + defer server.Close() + var conn net.Conn + var err error + if useTLS { + conn, err = tls.Dial("tcp", server.Listener.Addr().String(), &tls.Config{InsecureSkipVerify: true}) + } else { + conn, err = net.Dial("tcp", server.Listener.Addr().String()) + } + require.NoError(t, err) + defer conn.Close() + require.NoError(t, conn.SetDeadline(time.Now().Add(2*time.Second))) + _, err = io.WriteString(conn, "GET / HTTP/1.1\r\nHost: dashboard\r\nX-Incomplete: ") + require.NoError(t, err) + _, err = io.ReadAll(conn) + if networkErr, ok := err.(net.Error); ok { + require.False(t, networkErr.Timeout(), "server did not close incomplete request") + } + }) + } +} + +func TestDashboardHeaderTimeoutDoesNotCancelHTTP2Stream(t *testing.T) { + for _, useTLS := range []bool{false, true} { + name := "h2c" + if useTLS { + name = "h2" + } + t.Run(name, func(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.ProtoMajor != 2 { + t.Errorf("expected HTTP/2, got %s", r.Proto) + } + require.NoError(t, http.NewResponseController(w).Flush()) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("long-lived request body: %v", err) + return + } + _, _ = w.Write(body) + }) + server := httptest.NewUnstartedServer(handler) + server.Config = newDashboardHTTPServer(handler) + server.Config.ReadHeaderTimeout = 250 * time.Millisecond + server.Config.Protocols = new(http.Protocols) + server.Config.Protocols.SetHTTP1(true) + server.Config.Protocols.SetHTTP2(true) + server.Config.Protocols.SetUnencryptedHTTP2(true) + server.EnableHTTP2 = true + if useTLS { + server.StartTLS() + } else { + server.Start() + } + defer server.Close() + transport := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, Protocols: new(http.Protocols)} + transport.Protocols.SetHTTP2(true) + transport.Protocols.SetUnencryptedHTTP2(true) + defer transport.CloseIdleConnections() + client := &http.Client{Transport: transport, Timeout: 10 * time.Second} + reader, writer := io.Pipe() + defer reader.Close() + defer writer.Close() + request, err := http.NewRequest(http.MethodPost, server.URL, reader) + require.NoError(t, err) + response, err := client.Do(request) + require.NoError(t, err) + defer response.Body.Close() + time.Sleep(4 * server.Config.ReadHeaderTimeout) + _, err = writer.Write([]byte("still streaming")) + require.NoError(t, err) + require.NoError(t, writer.Close()) + body, err := io.ReadAll(response.Body) + require.NoError(t, err) + require.Equal(t, "still streaming", string(body)) + }) + } +} diff --git a/cmd/dashboard/main.go b/cmd/dashboard/main.go index 4c41ae0a..d636dc4c 100644 --- a/cmd/dashboard/main.go +++ b/cmd/dashboard/main.go @@ -12,7 +12,6 @@ import ( "os" "runtime/debug" "strings" - "time" _ "time/tzdata" "github.com/gin-gonic/gin" @@ -78,6 +77,16 @@ func initSystem(bus chan<- *model.Service) error { return err } + // 在线会话兜底清扫：状态流异常断开（失联）时由这里关闭进行中的会话 + if _, err := singleton.CronShared.AddFunc("@every 30s", singleton.SweepOnlineSessions); err != nil { + return err + } + + // 在线会话保留清理：超出后台设置的统计窗口后删除 + if _, err := singleton.CronShared.AddFunc("0 30 4 * * *", singleton.PruneOnlineSessions); err != nil { + return err + } + if err := singleton.StartJWTSessionGC(); err != nil { return err } @@ -172,10 +181,8 @@ func main() { wireNATDashboardCredentialGate() muxHandler := newHTTPandGRPCMux(httpHandler, grpcHandler) - muxServerHTTP := &http.Server{ - Handler: muxHandler, - ReadHeaderTimeout: time.Second * 5, - } + muxServerHTTP := newDashboardHTTPServer(muxHandler) + muxServerHTTP.Protocols = new(http.Protocols) muxServerHTTP.Protocols.SetHTTP1(true) muxServerHTTP.Protocols.SetUnencryptedHTTP2(true) @@ -183,9 +190,9 @@ func main() { var muxServerHTTPS *http.Server if singleton.Conf.HTTPS.ListenPort != 0 { muxServerHTTPS = &http.Server{ + ReadHeaderTimeout: dashboardReadHeaderTimeout, Addr: fmt.Sprintf("%s:%d", singleton.Conf.ListenHost, singleton.Conf.HTTPS.ListenPort), Handler: muxHandler, - ReadHeaderTimeout: time.Second * 5, TLSConfig: &tls.Config{ InsecureSkipVerify: singleton.Conf.HTTPS.InsecureTLS, }, diff --git a/model/config.go b/model/config.go index fde4bd19..6dc42b89 100644 --- a/model/config.go +++ b/model/config.go @@ -79,6 +79,10 @@ type ConfigDashboard struct { IgnoredIPNotification string `koanf:"ignored_ip_notification" json:"ignored_ip_notification,omitempty"` // 特定服务器IP（多个服务器用逗号分隔） DNSServers string `koanf:"dns_servers" json:"dns_servers,omitempty"` + + // OnlineStatsDays 设备在线时间段统计的保留与查询窗口（天）。 + // 取值范围 [MinOnlineStatsDays, MaxOnlineStatsDays]，默认 DefaultOnlineStatsDays。 + OnlineStatsDays int `koanf:"online_stats_days" json:"online_stats_days,omitempty"` } type Config struct { @@ -232,6 +236,13 @@ func (c *Config) Read(path string, frontendTemplates []FrontendTemplate) error { c.JWTTimeout = 1 } + if c.OnlineStatsDays == 0 { + c.OnlineStatsDays = DefaultOnlineStatsDays + } else if c.OnlineStatsDays < MinOnlineStatsDays || c.OnlineStatsDays > MaxOnlineStatsDays { + log.Printf("NEZHA>> online_stats_days %d out of range [%d, %d], reset to %d", c.OnlineStatsDays, MinOnlineStatsDays, MaxOnlineStatsDays, DefaultOnlineStatsDays) + c.OnlineStatsDays = DefaultOnlineStatsDays + } + c.jwtIPChangeAllowed.Store(c.AllowJWTIPChange) if c.AgentSecretKey == "" { diff --git a/model/online_history.go b/model/online_history.go new file mode 100644 index 00000000..79ba766c --- /dev/null +++ b/model/online_history.go @@ -0,0 +1,211 @@ +package model + +import ( + "gorm.io/gorm" + "sort" + "time" +) + +// 在线统计窗口的取值边界（天）。后台管理端可设置统计范围，查询窗口不允许 +// 超过该设置，历史数据按该设置做保留清理。 +const ( + DefaultOnlineStatsDays = 7 + MinOnlineStatsDays = 7 + MaxOnlineStatsDays = 3650 +) + +// OnlineActiveWindow 服务器状态上报距该窗口内视为在线。与 MCP server.list +// 的 onlineWindow 口径保持一致；超过该窗口未上报的进行中会话由定时清扫关闭。 +const OnlineActiveWindow = 30 * time.Second + +// OnlineSession 记录一台服务器的一段连续在线时间。Ongoing=true 表示会话 +// 仍在进行（此时 EndedAt 为零值）；会话结束时写入 EndedAt 并置 Ongoing=false。 +// LastActiveAt 由上报侧节流持久化，用于 dashboard 异常停机后把进行中的会话 +// 结转到最后一次已知活跃时刻。 +type OnlineSession struct { + ID uint64 `gorm:"primaryKey" json:"id,omitempty"` + CreatedAt time.Time `gorm:"index;<-:create" json:"created_at,omitempty"` + UpdatedAt time.Time `gorm:"autoUpdateTime" json:"updated_at,omitempty"` + ServerID uint64 `gorm:"index:idx_online_sessions_server_started;index:idx_online_sessions_server_closed,priority:1;index:idx_online_sessions_server_open,priority:1" json:"server_id,omitempty"` + StartedAt time.Time `gorm:"index:idx_online_sessions_server_started;index:idx_online_sessions_server_open,priority:3" json:"started_at,omitempty"` + EndedAt time.Time `gorm:"index;index:idx_online_sessions_server_closed,priority:3" json:"ended_at,omitempty"` // 会话结束时刻，进行中为零值 + // 上报侧节流持久化的最后活跃时刻（≥OnlineActiveWindow 间隔写一次） + LastActiveAt time.Time `json:"last_active_at,omitempty"` + Ongoing bool `gorm:"index:idx_online_sessions_server_closed,priority:2;index:idx_online_sessions_server_open,priority:2" json:"ongoing"` +} + +// BeforeSave keeps SQLite's lexicographic timestamp comparisons chronological. +// Query parameters and map-based updates must use UTC as well. +func (s *OnlineSession) BeforeSave(_ *gorm.DB) error { + s.StartedAt = s.StartedAt.UTC() + s.EndedAt = s.EndedAt.UTC() + s.LastActiveAt = s.LastActiveAt.UTC() + return nil +} + +// OnlinePeriod 是裁剪到查询窗口后的在线时间段，供前端时间轴直接渲染。 +type OnlinePeriod struct { + Start time.Time `json:"start"` + End time.Time `json:"end"` + Ongoing bool `json:"ongoing"` +} + +// DailyOnlineStat 是单个自然日（面板时区）在查询窗口内的在线秒数。 +// 窗口首尾为非整日时，只统计与窗口重叠的部分。 +type DailyOnlineStat struct { + Date string `json:"date"` + OnlineSeconds uint64 `json:"online_seconds"` +} + +// OnlineHistoryResponse 是单台服务器在线时间段统计的聚合结果。 +type OnlineHistoryResponse struct { + ServerID uint64 `json:"server_id"` + ServerName string `json:"server_name"` + Days int `json:"days"` + MaxDays int `json:"max_days"` + // Daily 拆日所用面板时区：IANA 名与窗口末端相对 UTC 的偏移秒数， + // 前端按同一时区渲染，保证跨时区访客的图形与数字一致 + Timezone string `json:"timezone"` + TzOffset int `json:"tz_offset"` + WindowStart time.Time `json:"window_start"` + WindowEnd time.Time `json:"window_end"` + TotalOnlineSeconds uint64 `json:"total_online_seconds"` + OnlineRate float64 `json:"online_rate"` + CurrentlyOnline bool `json:"currently_online"` + // 从未上报过的服务器该值为 nil；指针避免把 Go 零值时间序列化成 0001-01-01 + LastActive *time.Time `json:"last_active,omitempty"` + Periods []OnlinePeriod `json:"periods"` + Daily []DailyOnlineStat `json:"daily"` +} + +// ClipOnlineSessions 把与窗口 [windowStart, windowEnd) 相交的会话裁剪成 +// OnlinePeriod 列表。进行中的会话以 windowEnd 为右边界（保留 Ongoing 标记）。 +// 输入需按 StartedAt 升序；输出与输入同序。 +func ClipOnlineSessions(sessions []OnlineSession, windowStart, windowEnd time.Time) []OnlinePeriod { + periods := make([]OnlinePeriod, 0, len(sessions)) + for _, s := range sessions { + start := s.StartedAt + if start.Before(windowStart) { + start = windowStart + } + end := s.EndedAt + ongoing := s.Ongoing || end.IsZero() + if ongoing || end.After(windowEnd) { + end = windowEnd + } + if !start.Before(end) { + // 与窗口零交集（理论上不会由查询条件选出）或空会话 + continue + } + periods = append(periods, OnlinePeriod{Start: start, End: end, Ongoing: ongoing}) + } + return periods +} + +// DailyOnlineBreakdown 把在线时间段按面板时区的自然日拆分，返回窗口覆盖的 +// 每一天的在线秒数（含在线为 0 秒的日期），按日期升序。 +// 取整口径：每天先累计原始时长（日界切分不损失精度），仅在每日汇总时向下 +// 取整一次；调用方以各日之和作为总时长，保证响应内部 total 与 daily[] 严格 +// 对账，不与"逐段截断再累加"的口径混用（亚秒会话会让后者系统性偏小）。 +// 时段二分定位起始日界后逐日推进，复杂度 O(P·log D + D)；单服务器的会话 +// 互不重叠，内层推进总量受窗口天数约束，长窗口（如 days=3650）下不会出现 +// 时段数×天数的双重循环。 +func DailyOnlineBreakdown(periods []OnlinePeriod, windowStart, windowEnd time.Time, loc *time.Location) []DailyOnlineStat { + first := windowStart.In(loc) + // Advance a timezone-independent calendar, not a normalized local midnight: + // midnight DST gaps can normalize into the previous day and repeat dates. + date := time.Date(first.Year(), first.Month(), first.Day(), 0, 0, 0, 0, time.UTC) + var days []DailyOnlineStat + var dayStarts, dayEnds []time.Time + for { + start := onlineCalendarBoundary(date, loc) + if !start.Before(windowEnd) { + break + } + nextDate := date.AddDate(0, 0, 1) + end := onlineCalendarBoundary(nextDate, loc) + if start.Before(end) { + dayStarts = append(dayStarts, start) + dayEnds = append(dayEnds, end) + days = append(days, DailyOnlineStat{Date: date.Format("2006-01-02")}) + } + date = nextDate + } + + if len(days) == 0 { + return days + } + + dayDurations := make([]time.Duration, len(days)) + for _, p := range periods { + // 该时段覆盖的第一天：不晚于时段开始的最后一个日界 + i := sort.Search(len(dayStarts), func(i int) bool { + return dayStarts[i].After(p.Start) + }) - 1 + if i < 0 { + i = 0 + } + // 逐日推进，累加该时段与每一天 [dayStart, nextDay) 的交集 + for ; i < len(dayStarts) && dayStarts[i].Before(p.End); i++ { + overlapStart := maxTime(p.Start, dayStarts[i]) + overlapEnd := minTime(p.End, dayEnds[i]) + if overlapStart.Before(overlapEnd) { + dayDurations[i] += overlapEnd.Sub(overlapStart) + } + } + } + for i := range days { + days[i].OnlineSeconds = uint64(dayDurations[i].Seconds()) + } + return days +} + +func maxTime(a, b time.Time) time.Time { + if a.After(b) { + return a + } + return b +} + +func minTime(a, b time.Time) time.Time { + if a.Before(b) { + return a + } + return b +} + +// A missing midnight (or an entirely skipped date) starts at the timezone +// transition, not at the previous date returned by time.Date normalization. +func onlineCalendarBoundary(date time.Time, loc *time.Location) time.Time { + boundary := time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, loc) + if boundary.Format("2006-01-02") < date.Format("2006-01-02") { + _, end := boundary.ZoneBounds() + if !end.IsZero() { + return end + } + } + // time.Date may choose the second midnight in a rollback. Recover the + // earlier occurrence using the offset on the preceding side of the transition. + zoneStart, _ := boundary.ZoneBounds() + if !zoneStart.IsZero() { + _, currentOffset := boundary.Zone() + _, previousOffset := zoneStart.Add(-time.Nanosecond).Zone() + if previousOffset > currentOffset { + earlier := boundary.Add(-time.Duration(previousOffset-currentOffset) * time.Second) + if earlier.Format("2006-01-02 15:04:05") == boundary.Format("2006-01-02 15:04:05") { + boundary = earlier + } + } + } + return boundary +} + +// OnlineHistoryWindowStart defines the shared query and retention calendar window. +func OnlineHistoryWindowStart(now time.Time, days int, loc *time.Location) time.Time { + // Calendar days follow the dashboard timezone, not the host timezone; + // DST days can contain 23 or 25 hours. Queries and pruning must agree. + if loc == nil { + loc = time.UTC + } + return now.In(loc).AddDate(0, 0, -days).UTC() +} diff --git a/model/online_history_test.go b/model/online_history_test.go new file mode 100644 index 00000000..5403ef98 --- /dev/null +++ b/model/online_history_test.go @@ -0,0 +1,220 @@ +package model + +import ( + "math/rand" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +var testZone = time.FixedZone("TEST", 8*3600) + +func ts(y int, mo time.Month, d, h, mi, s int) time.Time { + return time.Date(y, mo, d, h, mi, s, 0, testZone) +} + +func TestClipOnlineSessions(t *testing.T) { + windowStart := ts(2026, 10, 1, 0, 0, 0) + windowEnd := ts(2026, 10, 8, 0, 0, 0) + + t.Run("session fully inside window", func(t *testing.T) { + sessions := []OnlineSession{ + {ServerID: 1, StartedAt: ts(2026, 10, 2, 8, 0, 0), EndedAt: ts(2026, 10, 2, 9, 30, 0)}, + } + periods := ClipOnlineSessions(sessions, windowStart, windowEnd) + require.Len(t, periods, 1) + require.Equal(t, ts(2026, 10, 2, 8, 0, 0), periods[0].Start) + require.Equal(t, ts(2026, 10, 2, 9, 30, 0), periods[0].End) + require.False(t, periods[0].Ongoing) + }) + + t.Run("session started before window is clipped", func(t *testing.T) { + sessions := []OnlineSession{ + {ServerID: 1, StartedAt: ts(2026, 9, 28, 12, 0, 0), EndedAt: ts(2026, 10, 1, 5, 0, 0)}, + } + periods := ClipOnlineSessions(sessions, windowStart, windowEnd) + require.Len(t, periods, 1) + require.Equal(t, windowStart, periods[0].Start) + require.Equal(t, ts(2026, 10, 1, 5, 0, 0), periods[0].End) + }) + + t.Run("ongoing session ends at window end", func(t *testing.T) { + sessions := []OnlineSession{ + {ServerID: 1, StartedAt: ts(2026, 10, 7, 10, 0, 0), Ongoing: true}, + } + periods := ClipOnlineSessions(sessions, windowStart, windowEnd) + require.Len(t, periods, 1) + require.Equal(t, windowEnd, periods[0].End) + require.True(t, periods[0].Ongoing) + }) + + t.Run("zero ended_at treated as ongoing", func(t *testing.T) { + sessions := []OnlineSession{ + {ServerID: 1, StartedAt: ts(2026, 10, 7, 10, 0, 0)}, + } + periods := ClipOnlineSessions(sessions, windowStart, windowEnd) + require.Len(t, periods, 1) + require.True(t, periods[0].Ongoing) + require.Equal(t, windowEnd, periods[0].End) + }) + + t.Run("session outside window yields no period", func(t *testing.T) { + sessions := []OnlineSession{ + {ServerID: 1, StartedAt: ts(2026, 9, 1, 0, 0, 0), EndedAt: ts(2026, 9, 2, 0, 0, 0)}, + } + require.Empty(t, ClipOnlineSessions(sessions, windowStart, windowEnd)) + }) +} + +func TestDailyOnlineBreakdown(t *testing.T) { + loc := time.FixedZone("UTC+8", 8*3600) + windowStart := time.Date(2026, 10, 1, 10, 0, 0, 0, loc) + windowEnd := time.Date(2026, 10, 3, 14, 0, 0, 0, loc) + + periods := []OnlinePeriod{ + // 跨午夜：2 日 23:00 ~ 3 日 01:00 + {Start: time.Date(2026, 10, 2, 23, 0, 0, 0, loc), End: time.Date(2026, 10, 3, 1, 0, 0, 0, loc)}, + // 窗口前就开始的会话，裁剪后从窗口起点开始 + {Start: windowStart, End: time.Date(2026, 10, 1, 12, 0, 0, 0, loc)}, + } + + daily := DailyOnlineBreakdown(periods, windowStart, windowEnd, loc) + require.Len(t, daily, 3) + require.Equal(t, "2026-10-01", daily[0].Date) + require.Equal(t, "2026-10-02", daily[1].Date) + require.Equal(t, "2026-10-03", daily[2].Date) + + // 1 日：12:00-10:00 = 2h；2 日：23:00-24:00 = 1h；3 日：0:00-1:00 + 窗口外无 = 1h + require.Equal(t, uint64(7200), daily[0].OnlineSeconds) + require.Equal(t, uint64(3600), daily[1].OnlineSeconds) + require.Equal(t, uint64(3600), daily[2].OnlineSeconds) + + // 每日合计即总时长（口径：每日累计原始时长后统一取整） + var sum uint64 + for _, d := range daily { + sum += d.OnlineSeconds + } + require.Equal(t, uint64(7200+3600+3600), sum) +} + +// 评审回归：取整必须按日累计原始时长后统一转换。逐段截断再累加会让亚秒 +// 会话系统性偏小——同一天两个 1.5s 的会话应得 3s，而不是 1+1=2s +func TestDailyOnlineBreakdownSubSecond(t *testing.T) { + loc := testZone + windowStart := ts(2026, 10, 1, 0, 0, 0) + windowEnd := ts(2026, 10, 2, 0, 0, 0) + periods := []OnlinePeriod{ + {Start: windowStart.Add(100 * time.Millisecond), End: windowStart.Add(1600 * time.Millisecond)}, + {Start: windowStart.Add(2 * time.Second), End: windowStart.Add(3500 * time.Millisecond)}, + } + daily := DailyOnlineBreakdown(periods, windowStart, windowEnd, loc) + require.Len(t, daily, 1) + require.Equal(t, uint64(3), daily[0].OnlineSeconds) +} + +// 跨日时段按日界切分不损失精度：各日分别累计原始时长后各自取整 +func TestDailyOnlineBreakdownCrossDayFraction(t *testing.T) { + loc := testZone + windowStart := ts(2026, 10, 1, 0, 0, 0) + windowEnd := ts(2026, 10, 3, 0, 0, 0) + periods := []OnlinePeriod{{ + Start: time.Date(2026, 10, 1, 23, 59, 59, 250000000, loc), + End: time.Date(2026, 10, 2, 0, 0, 2, 750000000, loc), + }} + daily := DailyOnlineBreakdown(periods, windowStart, windowEnd, loc) + require.Len(t, daily, 2) + require.Equal(t, uint64(0), daily[0].OnlineSeconds) // 0.75s → 0 + require.Equal(t, uint64(2), daily[1].OnlineSeconds) // 2.75s → 2 +} + +func TestDailyOnlineBreakdownZeroOnline(t *testing.T) { + loc := time.FixedZone("UTC+8", 8*3600) + windowStart := time.Date(2026, 10, 1, 0, 0, 0, 0, loc) + windowEnd := time.Date(2026, 10, 4, 0, 0, 0, 0, loc) + daily := DailyOnlineBreakdown(nil, windowStart, windowEnd, loc) + require.Len(t, daily, 3) + for _, d := range daily { + require.Zero(t, d.OnlineSeconds) + } +} + +func TestOnlineStatsDaysBounds(t *testing.T) { + require.Equal(t, 7, DefaultOnlineStatsDays) + require.Equal(t, 7, MinOnlineStatsDays) + require.Equal(t, 3650, MaxOnlineStatsDays) +} + +// F1 回归保护：指针推进实现必须与逐日朴素实现对任意不相交时段集合等价 +// （单服务器会话互不重叠的约束一致），长窗口大量时段下不再随时段数平方放大。 +func TestDailyOnlineBreakdownEquivalence(t *testing.T) { + loc := testZone + windowStart := ts(2026, 1, 1, 0, 0, 0) + windowEnd := windowStart.AddDate(0, 0, 90) + + rng := rand.New(rand.NewSource(42)) + var periods []OnlinePeriod + cursor := windowStart + for cursor.Before(windowEnd) { + // 起止带亚秒抖动：让逐段截断与按日累计两种口径可区分 + start := cursor.Add(time.Duration(rng.Intn(4*3600))*time.Second + time.Duration(rng.Int63n(int64(time.Second)))) + end := start.Add(time.Duration(1+rng.Intn(6*3600))*time.Second + time.Duration(rng.Int63n(int64(time.Second)))) + if end.After(windowEnd) { + end = windowEnd + } + periods = append(periods, OnlinePeriod{Start: start, End: end}) + cursor = end.Add(time.Duration(1+rng.Intn(3600)) * time.Second) + } + + days := DailyOnlineBreakdown(periods, windowStart, windowEnd, loc) + + // 参考实现与生产同一口径：每日累计原始时长、汇总时向下取整一次 + reference := map[string]time.Duration{} + for _, p := range periods { + for d := windowStart; d.Before(windowEnd); d = d.AddDate(0, 0, 1) { + overlapStart := maxTime(p.Start, d) + overlapEnd := minTime(p.End, d.AddDate(0, 0, 1)) + if overlapStart.Before(overlapEnd) { + reference[d.Format("2006-01-02")] += overlapEnd.Sub(overlapStart) + } + } + } + require.Len(t, days, 90) + for _, day := range days { + require.Equal(t, uint64(reference[day.Date].Seconds()), day.OnlineSeconds, "date %s", day.Date) + } +} + +func TestDailyOnlineBreakdownMidnightDSTTransition(t *testing.T) { + loc, err := time.LoadLocation("America/Sao_Paulo") + require.NoError(t, err) + start := time.Date(2018, 11, 3, 0, 0, 0, 0, loc) + end := time.Date(2018, 11, 6, 0, 0, 0, 0, loc) + days := DailyOnlineBreakdown([]OnlinePeriod{{Start: start, End: end}}, start, end, loc) + require.Equal(t, []DailyOnlineStat{{Date: "2018-11-03", OnlineSeconds: 86400}, {Date: "2018-11-04", OnlineSeconds: 82800}, {Date: "2018-11-05", OnlineSeconds: 86400}}, days) +} + +func TestDailyOnlineBreakdownRepeatedMidnight(t *testing.T) { + loc, err := time.LoadLocation("Asia/Amman") + require.NoError(t, err) + start := time.Date(2021, 10, 28, 21, 0, 0, 0, time.UTC) + end := time.Date(2021, 10, 29, 22, 0, 0, 0, time.UTC) + days := DailyOnlineBreakdown([]OnlinePeriod{{Start: start, End: end}}, start, end, loc) + require.Equal(t, []DailyOnlineStat{{Date: "2021-10-29", OnlineSeconds: 90000}}, days) +} + +func TestOnlineHistoryWindowUsesDashboardCalendar(t *testing.T) { + loc, err := time.LoadLocation("America/New_York") + require.NoError(t, err) + for _, test := range []struct { + month time.Month + day, hours int + }{{time.March, 9, 23}, {time.November, 2, 25}} { + end := time.Date(2026, test.month, test.day, 0, 30, 0, 0, loc) + start := OnlineHistoryWindowStart(end.UTC(), 1, loc) + require.True(t, start.Equal(time.Date(2026, test.month, test.day-1, 0, 30, 0, 0, loc))) + require.Equal(t, time.Duration(test.hours)*time.Hour, end.Sub(start)) + require.Len(t, DailyOnlineBreakdown(nil, start, end, loc), 2) + require.True(t, OnlineHistoryWindowStart(end.UTC(), 7, loc).Equal(end.In(loc).AddDate(0, 0, -7))) + } +} diff --git a/model/server.go b/model/server.go index ada61db9..3783ffe8 100644 --- a/model/server.go +++ b/model/server.go @@ -211,10 +211,16 @@ func (lease StateStreamLease) updateState(receiver *Server, state *HostState, la } func (lease StateStreamLease) Clear() bool { - return lease.clear(nil) + return lease.clear(nil, nil) } -func (lease StateStreamLease) clear(receiver *Server) bool { +// ClearWithSideEffect serializes disconnect persistence with replacement streams. +// The callback must not acquire the runtime holder again. +func (lease StateStreamLease) ClearWithSideEffect(sideEffect func()) bool { + return lease.clear(nil, sideEffect) +} + +func (lease StateStreamLease) clear(receiver *Server, sideEffect func()) bool { if lease.holder == nil { return false } @@ -226,6 +232,9 @@ func (lease StateStreamLease) clear(receiver *Server) bool { lease.holder.stream = nil lease.holder.lastActive = time.Time{} lease.holder.canonical.LastActive = time.Time{} + if sideEffect != nil { + sideEffect() + } return true } @@ -329,7 +338,7 @@ func (s *Server) UpdateStateIfCurrentWithSideEffect(lease StateStreamLease, stat } func (s *Server) ClearStateStreamIfCurrent(lease StateStreamLease) bool { - return lease.clear(s) + return lease.clear(s, nil) } // RuntimeSnapshot is a deep copy of the mutable runtime state. diff --git a/model/server_runtime_ownership_test.go b/model/server_runtime_ownership_test.go index 970ac7df..b29f5146 100644 --- a/model/server_runtime_ownership_test.go +++ b/model/server_runtime_ownership_test.go @@ -367,3 +367,30 @@ func TestServerRuntimeOwnership_failedRestartAllowsHourlyRecordThenRetry(t *test require.Equal(t, uint64(0), result.Transfer.In) require.Equal(t, uint64(0), result.Transfer.Out) } + +func TestStateStreamDisconnectSideEffectSerializesReplacement(t *testing.T) { + server := &Server{} + old := server.AttachStateStream(runtimeOwnershipStream{}) + entered := make(chan struct{}) + release := make(chan struct{}) + cleared := make(chan bool, 1) + go func() { cleared <- old.ClearWithSideEffect(func() { close(entered); <-release }) }() + <-entered + attached := make(chan StateStreamLease, 1) + go func() { attached <- server.AttachStateStream(runtimeOwnershipStream{}) }() + select { + case <-attached: + close(release) + t.Fatal("replacement attached before disconnect persistence finished") + case <-time.After(20 * time.Millisecond): + } + close(release) + require.True(t, <-cleared) + next := <-attached + require.True(t, next.UpdateState(&HostState{}, time.Now())) + called := false + require.False(t, old.ClearWithSideEffect(func() { called = true })) + require.False(t, called) + require.False(t, old.UpdateStateWithSideEffect(&HostState{}, time.Now(), func() error { called = true; return nil })) + require.False(t, called) +} diff --git a/model/setting_api.go b/model/setting_api.go index bd0510a0..0a530f54 100644 --- a/model/setting_api.go +++ b/model/setting_api.go @@ -21,6 +21,8 @@ type SettingForm struct { EnablePlainIPInNotification bool `json:"enable_plain_ip_in_notification,omitempty" validate:"optional"` AllowJWTIPChange *bool `json:"allow_jwt_ip_change,omitempty" validate:"optional"` EnableMCP *bool `json:"enable_mcp,omitempty" validate:"optional"` + + OnlineStatsDays *int `json:"online_stats_days,omitempty" validate:"optional"` // 在线统计范围（天），nil=保留现值 } type Setting struct { diff --git a/service/rpc/nezha.go b/service/rpc/nezha.go index e332aa74..30308c46 100644 --- a/service/rpc/nezha.go +++ b/service/rpc/nezha.go @@ -202,7 +202,15 @@ func (s *NezhaHandler) ReportSystemState(stream pb.NezhaService_ReportSystemStat return errors.New("server not found") } lease := server.AttachStateStream(stream) - defer lease.Clear() + defer func() { + // Persist while owning the lease; a replaced stream must never close a + // newer stream's session between Clear and RecordOffline. + lease.ClearWithSideEffect(func() { + if singleton.OnlineHistoryShared != nil { + singleton.OnlineHistoryShared.RecordOffline(clientID, time.Now()) + } + }) + }() var state *pb.State var stateCount uint64 for { @@ -253,6 +261,9 @@ func (s *NezhaHandler) ReportSystemState(stream pb.NezhaService_ReportSystemStat log.Printf("NEZHA>> Failed to write server metrics to TSDB: %v", err) } } + if singleton.OnlineHistoryShared != nil { + singleton.OnlineHistoryShared.RecordOnline(clientID, lastActive) + } return nil }) if !accepted { diff --git a/service/singleton/config.go b/service/singleton/config.go index 65f19687..6dcd1072 100644 --- a/service/singleton/config.go +++ b/service/singleton/config.go @@ -4,6 +4,7 @@ import ( "log" "strconv" "strings" + "sync" "github.com/nezhahq/nezha/model" "github.com/nezhahq/nezha/pkg/utils" @@ -14,6 +15,8 @@ var Conf *ConfigClass type ConfigClass struct { *model.Config + onlineHistoryRetentionMu sync.RWMutex + IgnoredIPNotificationServerIDs map[uint64]bool `json:"ignored_ip_notification_server_ids,omitempty"` Oauth2Providers []string `json:"oauth2_providers,omitempty"` } @@ -59,3 +62,39 @@ func (c *ConfigClass) updateIgnoredIPNotificationID() { } } } + +// OnlineHistoryRetentionDays returns the retention policy visible to queries and pruning. +func (c *ConfigClass) OnlineHistoryRetentionDays() int { + c.onlineHistoryRetentionMu.RLock() + defer c.onlineHistoryRetentionMu.RUnlock() + return c.OnlineStatsDays +} + +// CommitOnlineHistoryRetention applies a retention change with the settings save. +func (c *ConfigClass) CommitOnlineHistoryRetention(days *int, persist func() error) error { + // Readers must not capture a shorter cutoff while Save can still fail. + // Serialize assignment, persistence and rollback with the policy snapshot. + c.onlineHistoryRetentionMu.Lock() + defer c.onlineHistoryRetentionMu.Unlock() + previous := c.OnlineStatsDays + if days != nil { + c.OnlineStatsDays = *days + } + if err := persist(); err != nil { + c.OnlineStatsDays = previous + return err + } + return nil +} + +// SettingsSnapshot copies response fields without copying the retention mutex. +func (c *ConfigClass) SettingsSnapshot() model.Setting { + c.onlineHistoryRetentionMu.RLock() + defer c.onlineHistoryRetentionMu.RUnlock() + return model.Setting{ + ConfigForGuests: c.ConfigForGuests, + ConfigDashboard: c.ConfigDashboard, + IgnoredIPNotificationServerIDs: c.IgnoredIPNotificationServerIDs, + Oauth2Providers: c.Oauth2Providers, + } +} diff --git a/service/singleton/online_history.go b/service/singleton/online_history.go new file mode 100644 index 00000000..9f61a864 --- /dev/null +++ b/service/singleton/online_history.go @@ -0,0 +1,348 @@ +package singleton + +import ( + "fmt" + "log" + "sort" + "sync" + "time" + + "github.com/nezhahq/nezha/model" +) + +// OnlineHistoryRecorder 维护每台服务器当前进行中的在线会话。会话的打开与 +// 关闭由 agent 状态流的建立/断开驱动（service/rpc），并由定时清扫兜底关闭 +// 失联（未正常断流）的会话；dashboard 重启时通过数据库中 Ongoing 会话恢复。 +type OnlineHistoryRecorder struct { + mu sync.Mutex + restored bool + open map[uint64]*onlineSessionRuntime + // closing 是关库失败后的待关闭状态：断线边界已被观察到，会话 ID 与 + // 结束时间在此固定、不可被后续上报覆盖；重试只为旧会话完成持久化， + // 重连不得把它续接成活跃会话（同服务器可能同时有多条待关闭会话）。 + closing map[uint64][]*onlineSessionClosing + // Preserve the first disconnect while startup reads are unavailable. + recoveryDisconnects map[uint64]time.Time +} + +type onlineSessionRuntime struct { + id uint64 + lastObserved time.Time + revision uint64 + // lastPersist 是 LastActiveAt 最近一次落库时间，用于节流写库。 + lastPersist time.Time +} + +type onlineSessionClosing struct { + id uint64 + // end 是已观察到的会话结束时刻，重试持久化时原样使用。 + end time.Time +} + +var OnlineHistoryShared *OnlineHistoryRecorder + +func NewOnlineHistoryRecorder() *OnlineHistoryRecorder { + r := &OnlineHistoryRecorder{ + open: make(map[uint64]*onlineSessionRuntime), + closing: make(map[uint64][]*onlineSessionClosing), + recoveryDisconnects: make(map[uint64]time.Time), + } + r.mu.Lock() + r.restoreOngoingSessions() + r.mu.Unlock() + return r +} + +// restoreOngoingSessions 每台服务器仅允许最新且仍在在线窗口内的会话续接， +// 其余会话按最后持久化的活跃时间结转。崩溃会丢失内存中的待关闭边界， +// 恢复时采用 LastActiveAt 保守结转，不推测未持久化的断线时间。 +func (r *OnlineHistoryRecorder) restoreOngoingSessions() bool { + if DB == nil { + return false + } + var ongoing []model.OnlineSession + if err := DB.Where("ongoing = ?", true). + Order("server_id ASC").Order("started_at DESC").Order("id DESC").Find(&ongoing).Error; err != nil { + log.Printf("NEZHA>> Failed to load ongoing online sessions: %v", err) + return false + } + now := time.Now() + seenServers := make(map[uint64]bool) + for _, s := range ongoing { + // 关闭失败后重连可留下同服务器多条 Ongoing 行；直接赋值 open 会覆盖 + // 旧 ID，导致其永久脱管。按开始时间（同时间按 ID）选最新行，其余 + // 必须关闭或进入 closing 重试，即使最新行本身已经过期也不能复活旧行。 + duplicate := seenServers[s.ServerID] + seenServers[s.ServerID] = true + disconnect, observedDisconnect := r.recoveryDisconnects[s.ServerID] + observedDisconnect = observedDisconnect && !duplicate && !disconnect.Before(s.StartedAt) + if observedDisconnect || duplicate || s.LastActiveAt.Before(now.Add(-model.OnlineActiveWindow)) { + end := s.LastActiveAt + if observedDisconnect && disconnect.After(end) { + end = disconnect + } + if end.Before(s.StartedAt) { + end = s.StartedAt + } + if err := closeOnlineSessionRow(s.ID, end); err != nil { + log.Printf("NEZHA>> Failed to close recovered online session %d: %v", s.ID, err) + // 关库失败不能跳过，否则会留下永不闭合、也无人重试的 + // Ongoing 行；进入待关闭状态交由清扫周期重试。过期或 + // 被更新会话替代的行均不得被重连续接为活跃会话 + r.closing[s.ServerID] = append(r.closing[s.ServerID], + &onlineSessionClosing{id: s.ID, end: end}) + } + continue + } + r.open[s.ServerID] = &onlineSessionRuntime{id: s.ID, lastPersist: s.LastActiveAt, lastObserved: s.LastActiveAt} + } + clear(r.recoveryDisconnects) + r.restored = true + return true +} + +// flushClosing 尽力重试关闭该服务器的待关闭会话：按入列时固定的结束时间 +// 持久化，成功才移除；失败的条目原样保留，结束时间不得被后续上报改写。 +func (r *OnlineHistoryRecorder) flushClosing(serverID uint64) { + pendings, ok := r.closing[serverID] + if !ok { + return + } + kept := pendings[:0] + for _, pc := range pendings { + if err := closeOnlineSessionRow(pc.id, pc.end); err != nil { + log.Printf("NEZHA>> Failed to retry closing online session %d: %v", pc.id, err) + kept = append(kept, pc) + } + } + if len(kept) == 0 { + delete(r.closing, serverID) + } else { + r.closing[serverID] = kept + } +} + +// RecordOnline 记录一次状态上报：无进行中会话时开启新会话（重启遗留且仍在 +// 窗口内的 Ongoing 会话直接续接）；已有会话时节流落库 LastActiveAt。 +func (r *OnlineHistoryRecorder) RecordOnline(serverID uint64, at time.Time) { + r.mu.Lock() + defer r.mu.Unlock() + // A failed startup read is not an empty history: retry before accepting reports. + if !r.restored && !r.restoreOngoingSessions() { + return + } + at = at.UTC() + r.flushClosing(serverID) + + // A report can arrive after expiry but before the next sweep. Close at the + // previous observation before overwriting it, preserving failed-close retries. + if rt := r.open[serverID]; rt != nil && at.Sub(rt.lastObserved) > model.OnlineActiveWindow { + r.recordOfflineLocked(serverID, rt.lastObserved) + } + if rt, ok := r.open[serverID]; ok { + rt.lastObserved = at + rt.revision++ + if at.Sub(rt.lastPersist) >= model.OnlineActiveWindow { + if err := DB.Model(&model.OnlineSession{}).Where("id = ?", rt.id).Update("last_active_at", at).Error; err != nil { + log.Printf("NEZHA>> Failed to persist online session last_active_at: %v", err) + } else { + // Only committed writes advance the throttle; failures retry on the next report. + rt.lastPersist = at + } + } + return + } + + // All ongoing rows are owned by open or closing after successful recovery. + + s := model.OnlineSession{ServerID: serverID, StartedAt: at, LastActiveAt: at, Ongoing: true} + if err := DB.Create(&s).Error; err != nil { + log.Printf("NEZHA>> Failed to create online session: %v", err) + return + } + r.open[serverID] = &onlineSessionRuntime{id: s.ID, lastPersist: at, lastObserved: at} +} + +// RecordOffline 关闭该服务器当前进行中的会话。at 是会话结束时刻：正常断流 +// 为当前时间，清扫/结转为最后一次已知活跃时间。 +// 关库失败时把会话转入待关闭状态而非保留为活跃条目：会话 ID 与已观察到的 +// 结束时刻（at 与 lastPersist 的较大者）就此固定，后续上报不得覆盖；重试 +// 只为旧会话完成持久化，重连则开启新会话——断线后的离线间隔不得计作在线。 +func (r *OnlineHistoryRecorder) RecordOffline(serverID uint64, at time.Time) { + r.mu.Lock() + defer r.mu.Unlock() + // Disconnect may be the first event after startup storage recovers. + if !r.restored && !r.restoreOngoingSessions() { + // Retrying SELECT alone loses confirmed offline gaps on a quick reconnect. + // Keep the first boundary; subsequent disconnected streams cannot extend it. + if r.recoveryDisconnects == nil { + r.recoveryDisconnects = make(map[uint64]time.Time) + } + if _, exists := r.recoveryDisconnects[serverID]; !exists { + r.recoveryDisconnects[serverID] = at.UTC() + } + return + } + r.recordOfflineLocked(serverID, at) +} + +func (r *OnlineHistoryRecorder) recordOfflineLocked(serverID uint64, at time.Time) { + rt, ok := r.open[serverID] + if !ok { + return + } + end := at + if end.Before(rt.lastPersist) { + end = rt.lastPersist + } + if err := closeOnlineSessionRow(rt.id, end); err != nil { + log.Printf("NEZHA>> Failed to close online session %d: %v", rt.id, err) + // 结束时刻已观察到，不能因写库失败丢失，也不能把行继续当活跃 + // 会话：转待关闭状态，由 RecordOnline 冲洗与清扫周期重试 + r.closing[serverID] = append(r.closing[serverID], &onlineSessionClosing{id: rt.id, end: end}) + delete(r.open, serverID) + return + } + delete(r.open, serverID) +} + +// SweepStale 关闭超过在线窗口仍无状态上报的进行中会话。lastActiveFn 返回 +// 服务器的运行时最后活跃时间与是否存在。 +func (r *OnlineHistoryRecorder) SweepStale(now time.Time, lastActiveFn func(serverID uint64) (time.Time, bool)) { + // Never call the runtime observer while holding mu: reports acquire the + // runtime lease before this mutex. Snapshot identities, then revalidate them. + r.mu.Lock() + if !r.restored && !r.restoreOngoingSessions() { + r.mu.Unlock() + return + } + candidates := make(map[uint64]onlineSessionRuntime, len(r.open)) + for id, rt := range r.open { + candidates[id] = *rt + } + for id := range r.closing { + r.flushClosing(id) + } + r.mu.Unlock() + for id, candidate := range candidates { + lastActive, ok := lastActiveFn(id) + r.mu.Lock() + rt := r.open[id] + if rt != nil && rt.id == candidate.id && rt.revision == candidate.revision { + if lastActive.Before(rt.lastObserved) { + lastActive = rt.lastObserved + } + if (!ok || lastActive.Before(now.Add(-model.OnlineActiveWindow))) && rt.lastObserved.Before(now.Add(-model.OnlineActiveWindow)) { + r.recordOfflineLocked(id, lastActive) + } + } + r.mu.Unlock() + } + +} + +// SweepOnlineSessions 是定时清扫入口，接上服务器运行时的最后活跃时间。 +func SweepOnlineSessions() { + if OnlineHistoryShared == nil || ServerShared == nil || DB == nil { + return + } + OnlineHistoryShared.SweepStale(time.Now(), onlineServerLastActive) +} + +func onlineServerLastActive(serverID uint64) (time.Time, bool) { + server, ok := ServerShared.Get(serverID) + if !ok || server == nil { + return time.Time{}, false + } + lastActive := server.RuntimeSnapshot().LastActive + return lastActive, !lastActive.IsZero() +} + +// PruneOnlineSessions 清理超出统计窗口的已结束会话与已删除服务器的残留记录。 +// 时间列直接比较（与写入时的绑定格式同构）以走 ended_at 索引，并用分批 +// DELETE 替换单条巨型删除，避免大表上长时间占用 sqlite 写锁、阻塞上报路径。 +func PruneOnlineSessions() { + if DB == nil { + return + } + keepDays := Conf.OnlineHistoryRetentionDays() + if keepDays < model.MinOnlineStatsDays || keepDays > model.MaxOnlineStatsDays { + keepDays = model.DefaultOnlineStatsDays + } + cutoff := model.OnlineHistoryWindowStart(time.Now(), keepDays, Loc) + pruneOnlineSessionsBatch("ongoing = ? AND ended_at < ?", false, cutoff) + pruneOnlineSessionsBatch("server_id NOT IN (SELECT `id` FROM servers)") +} + +// pruneOnlineSessionsBatch 按主键分批删除匹配的行，每批之间释放写锁。 +func pruneOnlineSessionsBatch(where string, args ...any) { + const batchSize = 10000 + for { + var ids []uint64 + if err := DB.Model(&model.OnlineSession{}).Unscoped(). + Where(where, args...).Limit(batchSize).Pluck("id", &ids).Error; err != nil { + log.Printf("NEZHA>> Failed to select online sessions for pruning: %v", err) + return + } + if len(ids) == 0 { + return + } + if err := DB.Unscoped().Delete(&model.OnlineSession{}, ids).Error; err != nil { + log.Printf("NEZHA>> Failed to prune online sessions: %v", err) + return + } + if len(ids) < batchSize { + return + } + } +} + +func closeOnlineSessionRow(id uint64, end time.Time) error { + return DB.Model(&model.OnlineSession{}).Where("id = ?", id).Updates(map[string]any{ + "ended_at": end.UTC(), + "ongoing": false, + }).Error +} + +// Sessions snapshots rows and volatile close boundaries under the same lock. +// Otherwise a failed close can count confirmed offline time as ongoing, or a +// successful retry can remove the boundary after the reader fetched a stale row. +func (r *OnlineHistoryRecorder) Sessions(serverID uint64, start, end time.Time) ([]model.OnlineSession, error) { + if r != nil { + r.mu.Lock() + defer r.mu.Unlock() + if !r.restored && !r.restoreOngoingSessions() { + return nil, fmt.Errorf("online history recovery unavailable") + } + } + var sessions []model.OnlineSession + // Separate branches use (server_id, ongoing, ended_at) and + // (server_id, ongoing, started_at), avoiding a scan of retained closed rows. + for _, ongoing := range []bool{false, true} { + var rows []model.OnlineSession + q := DB.Where("server_id = ? AND ongoing = ?", serverID, ongoing) + if ongoing { + q = q.Where("started_at < ?", end.UTC()) + } else { + q = q.Where("ended_at > ?", start.UTC()) + } + if err := q.Find(&rows).Error; err != nil { + return nil, err + } + sessions = append(sessions, rows...) + } + if r != nil { + endings := make(map[uint64]time.Time) + for _, pending := range r.closing[serverID] { + endings[pending.id] = pending.end + } + for i := range sessions { + if end, ok := endings[sessions[i].ID]; ok { + sessions[i].Ongoing = false + sessions[i].EndedAt = end + } + } + } + sort.Slice(sessions, func(i, j int) bool { return sessions[i].StartedAt.Before(sessions[j].StartedAt) }) + return sessions, nil +} diff --git a/service/singleton/online_history_failure_test.go b/service/singleton/online_history_failure_test.go new file mode 100644 index 00000000..8f22a7e8 --- /dev/null +++ b/service/singleton/online_history_failure_test.go @@ -0,0 +1,224 @@ +package singleton + +import ( + "errors" + "testing" + "time" + + "github.com/nezhahq/nezha/model" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +func TestOnlineHistoryRetriesFailedHeartbeatImmediately(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + start := time.Now().UTC().Truncate(time.Second) + r.RecordOnline(1, start) + id := r.open[1].id + require.NoError(t, DB.Exec(`CREATE TRIGGER reject_heartbeat BEFORE UPDATE OF last_active_at ON online_sessions BEGIN SELECT RAISE(ABORT, 'heartbeat failure'); END`).Error) + r.RecordOnline(1, start.Add(30*time.Second)) + require.NoError(t, DB.Exec("DROP TRIGGER reject_heartbeat").Error) + r.RecordOnline(1, start.Add(31*time.Second)) + require.True(t, getOnlineSession(t, id).LastActiveAt.Equal(start.Add(31*time.Second))) +} + +func TestOnlineHistoryRetriesStartupReadBeforeNewReport(t *testing.T) { + setupOnlineHistoryTestDB(t) + start := time.Now().UTC().Add(-time.Hour).Truncate(time.Second) + old := model.OnlineSession{ServerID: 1, StartedAt: start, LastActiveAt: start, Ongoing: true} + require.NoError(t, DB.Create(&old).Error) + require.NoError(t, DB.Callback().Query().Before("gorm:query").Register("reject_history_restore", func(db *gorm.DB) { db.AddError(errors.New("restore unavailable")) })) + r := NewOnlineHistoryRecorder() + require.NoError(t, DB.Callback().Query().Remove("reject_history_restore")) + r.RecordOnline(1, time.Now()) + stored := getOnlineSession(t, old.ID) + require.False(t, stored.Ongoing) + require.True(t, stored.EndedAt.Equal(start)) + require.NotEqual(t, old.ID, r.open[1].id) + require.Equal(t, int64(1), countOnlineSessions(t, "ongoing = ?", true)) +} + +func TestOnlineHistorySweepKeepsSessionReconnectedDuringObservation(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + start := time.Now().Add(-time.Hour) + r.RecordOnline(1, start) + observed := make(chan struct{}) + release := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + r.SweepStale(time.Now(), func(uint64) (time.Time, bool) { close(observed); <-release; return start, true }) + }() + <-observed + reconnected := make(chan struct{}) + go func() { r.RecordOffline(1, start.Add(time.Second)); r.RecordOnline(1, time.Now()); close(reconnected) }() + // The observer must not hold the recorder lock: RPC updates take the runtime + // lock first and then enter the recorder. Holding both in reverse order deadlocks. + select { + case <-reconnected: + case <-time.After(time.Second): + close(release) + <-done + <-reconnected + t.Fatal("sweep holds recorder lock while consulting runtime state") + } + id := r.open[1].id + close(release) + <-done + require.Contains(t, r.open, uint64(1)) + require.Equal(t, id, r.open[1].id) + require.True(t, getOnlineSession(t, id).Ongoing) +} + +func TestOnlineHistoryQueryNormalizesOffsetsAndUsesWindowIndexes(t *testing.T) { + setupOnlineHistoryTestDB(t) + east := time.FixedZone("UTC+8", 8*3600) + start := time.Date(2025, 1, 1, 12, 0, 0, 0, east) + r := NewOnlineHistoryRecorder() + r.RecordOnline(1, start) + r.RecordOffline(1, start.Add(time.Hour)) + rows, err := r.Sessions(1, start.UTC().Add(-time.Minute), start.UTC().Add(2*time.Hour)) + require.NoError(t, err) + require.Len(t, rows, 1) + require.True(t, rows[0].StartedAt.Equal(start)) + require.True(t, rows[0].EndedAt.Equal(start.Add(time.Hour))) + for _, test := range []struct { + query, index string + args []any + }{ + {"EXPLAIN QUERY PLAN SELECT * FROM online_sessions WHERE server_id = ? AND ongoing = ? AND ended_at > ?", "idx_online_sessions_server_closed", []any{1, false, start.UTC()}}, + {"EXPLAIN QUERY PLAN SELECT * FROM online_sessions WHERE server_id = ? AND ongoing = ? AND started_at < ?", "idx_online_sessions_server_open", []any{1, true, start.Add(2 * time.Hour).UTC()}}, + } { + var plan []struct{ Detail string } + require.NoError(t, DB.Raw(test.query, test.args...).Scan(&plan).Error) + require.NotEmpty(t, plan) + require.Contains(t, plan[0].Detail, test.index) + } +} + +func TestOnlineHistorySweepKeepsReportReceivedDuringObservation(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + start := time.Now().Add(-time.Hour) + r.RecordOnline(1, start) + id := r.open[1].id + r.SweepStale(time.Now(), func(uint64) (time.Time, bool) { + r.RecordOnline(1, time.Now()) + return start, true + }) + require.Contains(t, r.open, uint64(1)) + // A report after an hour starts a new session; the stale sweep must not close it. + require.NotEqual(t, id, r.open[1].id) + require.False(t, getOnlineSession(t, id).Ongoing) + require.True(t, getOnlineSession(t, r.open[1].id).Ongoing) +} + +func TestOnlineHistoryStartupReadFailureBlocksQueryAndSweepRetries(t *testing.T) { + setupOnlineHistoryTestDB(t) + start := time.Now().Add(-time.Hour) + old := model.OnlineSession{ServerID: 1, StartedAt: start, LastActiveAt: start, Ongoing: true} + require.NoError(t, DB.Create(&old).Error) + require.NoError(t, DB.Callback().Query().Before("gorm:query").Register("reject_history_restore", func(db *gorm.DB) { db.AddError(errors.New("restore unavailable")) })) + r := NewOnlineHistoryRecorder() + _, err := r.Sessions(1, start, time.Now()) + require.ErrorContains(t, err, "recovery unavailable") + r.RecordOnline(1, time.Now()) + require.Empty(t, r.open) + require.NoError(t, DB.Callback().Query().Remove("reject_history_restore")) + r.SweepStale(time.Now(), func(uint64) (time.Time, bool) { return time.Time{}, false }) + require.False(t, getOnlineSession(t, old.ID).Ongoing) + require.Equal(t, int64(0), countOnlineSessions(t, "ongoing = ?", true)) +} + +func TestOnlineHistoryDisconnectRetriesStartupRead(t *testing.T) { + setupOnlineHistoryTestDB(t) + start := time.Now().Add(-10 * time.Second) + old := model.OnlineSession{ServerID: 1, StartedAt: start, LastActiveAt: start, Ongoing: true} + require.NoError(t, DB.Create(&old).Error) + require.NoError(t, DB.Callback().Query().Before("gorm:query").Register("reject_history_restore", func(db *gorm.DB) { db.AddError(errors.New("restore unavailable")) })) + r := NewOnlineHistoryRecorder() + require.NoError(t, DB.Callback().Query().Remove("reject_history_restore")) + end := start.Add(time.Second) + r.RecordOffline(1, end) + require.False(t, getOnlineSession(t, old.ID).Ongoing) + require.True(t, getOnlineSession(t, old.ID).EndedAt.Equal(end)) + r.RecordOnline(1, time.Now()) + require.NotEqual(t, old.ID, r.open[1].id) +} + +func TestOnlineHistoryReportAfterTimeoutSplitsBeforeSweep(t *testing.T) { + for _, failClose := range []bool{false, true} { + t.Run(map[bool]string{false: "persisted", true: "retry_close"}[failClose], func(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + start := time.Now().UTC().Truncate(time.Second) + r.RecordOnline(1, start) + r.RecordOnline(1, start.Add(time.Second)) + first := r.open[1].id + r.SweepStale(start.Add(30*time.Second), func(uint64) (time.Time, bool) { return start.Add(time.Second), true }) + require.Equal(t, first, r.open[1].id) + if failClose { + injectCloseUpdateFailure(t) + } + r.RecordOnline(1, start.Add(45*time.Second)) + require.NotEqual(t, first, r.open[1].id) + second := r.open[1].id + rows, err := r.Sessions(1, start, start.Add(46*time.Second)) + require.NoError(t, err) + require.Len(t, rows, 2) + require.False(t, rows[0].Ongoing) + require.True(t, rows[0].EndedAt.Equal(start.Add(time.Second))) + require.True(t, rows[1].StartedAt.Equal(start.Add(45*time.Second))) + if failClose { + require.Len(t, r.closing[1], 1) + dropCloseUpdateFailure(t) + } + r.SweepStale(start.Add(46*time.Second), func(uint64) (time.Time, bool) { return start.Add(45 * time.Second), true }) + require.Empty(t, r.closing) + require.False(t, getOnlineSession(t, first).Ongoing) + require.True(t, getOnlineSession(t, first).EndedAt.Equal(start.Add(time.Second))) + require.True(t, getOnlineSession(t, second).Ongoing) + }) + } +} + +func TestOnlineHistoryDisconnectSurvivesUnavailableRecovery(t *testing.T) { + for _, failClose := range []bool{false, true} { + t.Run(map[bool]string{false: "close_success", true: "close_retry"}[failClose], func(t *testing.T) { + setupOnlineHistoryTestDB(t) + start := time.Now().UTC().Add(-10 * time.Second) + old := model.OnlineSession{ServerID: 1, StartedAt: start, LastActiveAt: start, Ongoing: true} + require.NoError(t, DB.Create(&old).Error) + require.NoError(t, DB.Callback().Query().Before("gorm:query").Register("unavailable_disconnect_restore", func(db *gorm.DB) { db.AddError(errors.New("restore unavailable")) })) + r := NewOnlineHistoryRecorder() + end := start.Add(time.Second) + r.RecordOffline(1, end) + // Later disconnects while recovery is unavailable must not extend the first session. + r.RecordOffline(1, end.Add(time.Second)) + require.NoError(t, DB.Callback().Query().Remove("unavailable_disconnect_restore")) + if failClose { + injectCloseUpdateFailure(t) + } + reconnect := start.Add(5 * time.Second) + r.RecordOnline(1, reconnect) + require.NotEqual(t, old.ID, r.open[1].id) + rows, err := r.Sessions(1, start, reconnect.Add(time.Second)) + require.NoError(t, err) + require.Len(t, rows, 2) + require.False(t, rows[0].Ongoing) + require.True(t, rows[0].EndedAt.Equal(end)) + require.True(t, rows[1].StartedAt.Equal(reconnect)) + if failClose { + require.Len(t, r.closing[1], 1) + dropCloseUpdateFailure(t) + } + r.SweepStale(reconnect.Add(time.Second), func(uint64) (time.Time, bool) { return reconnect, true }) + require.Empty(t, r.closing) + require.False(t, getOnlineSession(t, old.ID).Ongoing) + require.True(t, getOnlineSession(t, old.ID).EndedAt.Equal(end)) + require.True(t, getOnlineSession(t, r.open[1].id).Ongoing) + }) + } +} diff --git a/service/singleton/online_history_restore_test.go b/service/singleton/online_history_restore_test.go new file mode 100644 index 00000000..ebafdaaf --- /dev/null +++ b/service/singleton/online_history_restore_test.go @@ -0,0 +1,73 @@ +package singleton + +import ( + "testing" + "time" + + "github.com/nezhahq/nezha/model" + "github.com/stretchr/testify/require" +) + +func TestRestoreOngoingSessionsReconcilesRecentDuplicates(t *testing.T) { + for _, failClose := range []bool{false, true} { + name := "successful_close" + if failClose { + name = "retry_failed_close" + } + t.Run(name, func(t *testing.T) { + setupOnlineHistoryTestDB(t) + base := time.Now().Add(-10 * time.Second).Truncate(time.Second) + // Insert out of chronological order: choosing by ID alone would keep the wrong session. + newest := model.OnlineSession{ServerID: 1, StartedAt: base.Add(4 * time.Second), LastActiveAt: base.Add(5 * time.Second), Ongoing: true} + oldest := model.OnlineSession{ServerID: 1, StartedAt: base, LastActiveAt: base.Add(time.Second), Ongoing: true} + middle := model.OnlineSession{ServerID: 1, StartedAt: base.Add(2 * time.Second), LastActiveAt: base.Add(3 * time.Second), Ongoing: true} + for _, session := range []*model.OnlineSession{&newest, &oldest, &middle} { + require.NoError(t, DB.Create(session).Error) + } + if failClose { + injectCloseUpdateFailure(t) + } + r := NewOnlineHistoryRecorder() + require.Len(t, r.open, 1) + require.Equal(t, newest.ID, r.open[1].id) + if failClose { + require.Len(t, r.closing[1], 2) + require.Equal(t, int64(3), countOnlineSessions(t, "ongoing = ?", true)) + dropCloseUpdateFailure(t) + } + r.SweepStale(base.Add(6*time.Second), func(uint64) (time.Time, bool) { return newest.LastActiveAt, true }) + require.Empty(t, r.closing) + require.True(t, getOnlineSession(t, newest.ID).Ongoing) + for _, previous := range []model.OnlineSession{oldest, middle} { + stored := getOnlineSession(t, previous.ID) + require.False(t, stored.Ongoing) + require.True(t, stored.EndedAt.Equal(previous.LastActiveAt)) + } + r.RecordOnline(1, base.Add(7*time.Second)) + require.Equal(t, newest.ID, r.open[1].id) + require.Equal(t, int64(3), countOnlineSessions(t, "server_id = ?", 1)) + r.SweepStale(base.Add(2*model.OnlineActiveWindow), func(uint64) (time.Time, bool) { return time.Time{}, false }) + require.Empty(t, r.open) + require.Empty(t, r.closing) + require.Equal(t, int64(0), countOnlineSessions(t, "ongoing = ?", true)) + }) + } +} + +func TestRestoreOngoingSessionsUsesIDToBreakStartTimeTies(t *testing.T) { + setupOnlineHistoryTestDB(t) + start := time.Now().Add(-5 * time.Second) + first := model.OnlineSession{ServerID: 1, StartedAt: start, LastActiveAt: start, Ongoing: true} + second := first + other := model.OnlineSession{ServerID: 2, StartedAt: start, LastActiveAt: start, Ongoing: true} + for _, session := range []*model.OnlineSession{&first, &second, &other} { + require.NoError(t, DB.Create(session).Error) + } + r := NewOnlineHistoryRecorder() + require.Len(t, r.open, 2) + require.Equal(t, second.ID, r.open[1].id) + require.Equal(t, other.ID, r.open[2].id) + require.False(t, getOnlineSession(t, first.ID).Ongoing) + require.True(t, getOnlineSession(t, first.ID).EndedAt.Equal(start)) + require.Equal(t, int64(2), countOnlineSessions(t, "ongoing = ?", true)) +} diff --git a/service/singleton/online_history_retention_test.go b/service/singleton/online_history_retention_test.go new file mode 100644 index 00000000..f505ba2f --- /dev/null +++ b/service/singleton/online_history_retention_test.go @@ -0,0 +1,63 @@ +package singleton + +import ( + "errors" + "testing" + "time" + + "github.com/nezhahq/nezha/model" + "github.com/stretchr/testify/require" +) + +func TestOnlineHistoryPruneWaitsForRetentionCommit(t *testing.T) { + for _, saveFails := range []bool{true, false} { + t.Run(map[bool]string{true: "failed_save", false: "successful_save"}[saveFails], func(t *testing.T) { + setupOnlineHistoryTestDB(t) + previous := Conf + Conf = &ConfigClass{Config: &model.Config{}} + Conf.OnlineStatsDays = 30 + t.Cleanup(func() { Conf = previous }) + now := time.Now().UTC() + row := model.OnlineSession{ServerID: 1, StartedAt: now.AddDate(0, 0, -15), EndedAt: now.AddDate(0, 0, -14)} + require.NoError(t, DB.Create(&row).Error) + saveEntered := make(chan struct{}) + releaseSave := make(chan struct{}) + saved := make(chan error, 1) + days := 7 + go func() { + saved <- Conf.CommitOnlineHistoryRetention(&days, func() error { + close(saveEntered) + <-releaseSave + if saveFails { + return errors.New("retention save unavailable") + } + return nil + }) + }() + <-saveEntered + pruneStarted := make(chan struct{}) + pruned := make(chan struct{}) + go func() { close(pruneStarted); PruneOnlineSessions(); close(pruned) }() + <-pruneStarted + premature := false + select { + case <-pruned: + premature = true + case <-time.After(100 * time.Millisecond): + } + close(releaseSave) + err := <-saved + <-pruned + require.False(t, premature, "pruning must not read an uncommitted retention policy") + if saveFails { + require.ErrorContains(t, err, "retention save unavailable") + require.Equal(t, 30, Conf.OnlineHistoryRetentionDays()) + require.Equal(t, int64(1), countOnlineSessions(t, "id = ?", row.ID)) + } else { + require.NoError(t, err) + require.Equal(t, 7, Conf.OnlineHistoryRetentionDays()) + require.Equal(t, int64(0), countOnlineSessions(t, "id = ?", row.ID)) + } + }) + } +} diff --git a/service/singleton/online_history_test.go b/service/singleton/online_history_test.go new file mode 100644 index 00000000..285f8f31 --- /dev/null +++ b/service/singleton/online_history_test.go @@ -0,0 +1,316 @@ +package singleton + +import ( + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + "gorm.io/gorm" + + "github.com/nezhahq/nezha/model" +) + +func setupOnlineHistoryTestDB(t *testing.T) { + t.Helper() + + previousDB := DB + var err error + DB, err = gorm.Open(openSQLiteDialector(filepath.Join(t.TempDir(), "dashboard.sqlite")), &gorm.Config{}) + require.NoError(t, err) + sqlDB, err := DB.DB() + require.NoError(t, err) + t.Cleanup(func() { + DB = previousDB + if err := sqlDB.Close(); err != nil { + t.Errorf("close online history test database: %v", err) + } + }) + + require.NoError(t, DB.AutoMigrate(&model.Server{}, &model.OnlineSession{})) + require.NoError(t, DB.Exec("INSERT INTO servers (id, name, uuid) VALUES (1, 'server-1', 'online-history-test-1')").Error) +} + +func getOnlineSession(t *testing.T, id uint64) model.OnlineSession { + t.Helper() + var s model.OnlineSession + require.NoError(t, DB.First(&s, id).Error) + return s +} + +func countOnlineSessions(t *testing.T, cond string, args ...any) int64 { + t.Helper() + var count int64 + require.NoError(t, DB.Model(&model.OnlineSession{}).Where(cond, args...).Count(&count).Error) + return count +} + +// injectCloseUpdateFailure 用 SQLite 触发器注入"SELECT 正常、仅关闭 UPDATE 失败"： +// 只拦截把 ongoing 置 0 的更新，LastActiveAt 节流写与 INSERT 均不受影响， +// 使后续 RecordOnline/恢复/清扫全部走真实执行路径。 +func injectCloseUpdateFailure(t *testing.T) { + t.Helper() + require.NoError(t, DB.Exec(`CREATE TRIGGER fail_online_session_close +BEFORE UPDATE ON online_sessions +WHEN NEW.ongoing = 0 +BEGIN + SELECT RAISE(ABORT, 'injected close failure'); +END`).Error) + t.Cleanup(func() { dropCloseUpdateFailure(t) }) +} + +func dropCloseUpdateFailure(t *testing.T) { + t.Helper() + require.NoError(t, DB.Exec("DROP TRIGGER IF EXISTS fail_online_session_close").Error) +} + +// 评审回归（P2：关库失败丢边界/重连续接待关闭会话）：RecordOffline 关库失败时 +// 必须转入待关闭状态——会话 ID 与已观察到的断线时刻固定、后续上报不得覆盖； +// 重连开启的是新会话而非续接旧行；失败恢复后由清扫周期按固定边界完成持久化， +// 离线间隔不计入在线。时间自单一基准单调推进。 +func TestRecordOfflineCloseFailurePreservesBoundary(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + + base := time.Now().Add(-time.Hour).Truncate(time.Second) + atStart := base + atLastActive := atStart.Add(model.OnlineActiveWindow) // 恰好达到节流窗口，触发落库 + atOffline := atLastActive.Add(15 * time.Second) // 观察到的断线时刻 + atReconnect := atOffline.Add(10 * time.Minute) + + r.RecordOnline(1, atStart) + r.RecordOnline(1, atLastActive) + require.Len(t, r.open, 1) + first := getOnlineSession(t, r.open[1].id) + require.WithinDuration(t, atLastActive, first.LastActiveAt, time.Second) + + // 注入关闭失败：行保持 Ongoing 等待重试，会话转入待关闭而非继续当活跃会话 + injectCloseUpdateFailure(t) + r.RecordOffline(1, atOffline) + require.NotContains(t, r.open, uint64(1)) + stuck := getOnlineSession(t, first.ID) + require.True(t, stuck.Ongoing) + require.True(t, stuck.EndedAt.IsZero(), "关库失败时旧行必须保持未闭合等待重试") + + // 失败未恢复时 agent 重连：旧行不得被续接，必须开启新会话 + r.RecordOnline(1, atReconnect) + require.Contains(t, r.open, uint64(1)) + require.NotEqual(t, first.ID, r.open[1].id, "待关闭会话不得被续接为活跃会话") + afterRetry := getOnlineSession(t, first.ID) + require.True(t, afterRetry.Ongoing, "失败未恢复时重试仍失败，旧行保持待关闭") + second := getOnlineSession(t, r.open[1].id) + require.True(t, second.Ongoing) + require.WithinDuration(t, atReconnect, second.StartedAt, time.Second) + + // 失败恢复后清扫：旧行按观察到的断线时刻闭合，新会话仍活跃不受影响 + dropCloseUpdateFailure(t) + r.SweepStale(atReconnect.Add(time.Second), func(uint64) (time.Time, bool) { + return atReconnect, true + }) + + closed := getOnlineSession(t, first.ID) + require.False(t, closed.Ongoing) + require.WithinDuration(t, atOffline, closed.EndedAt, time.Second, "旧行必须按观察到的断线时刻闭合") + require.WithinDuration(t, atStart, closed.StartedAt, time.Second) + stillOngoing := getOnlineSession(t, second.ID) + require.True(t, stillOngoing.Ongoing) + + // 离线间隔不计入在线：两行在 [atStart, atReconnect) 内的在线合计止于断线时刻 + var rows []model.OnlineSession + require.NoError(t, DB.Where("server_id = ?", 1).Order("id").Find(&rows).Error) + require.Len(t, rows, 2) + var online time.Duration + for _, s := range rows { + end := s.EndedAt + if s.Ongoing { + end = atReconnect + } + online += end.Sub(s.StartedAt) + } + require.Equal(t, atOffline.Sub(atStart), online, "离线间隔不得计作在线") +} + +// 评审回归（P2 启动恢复路径）：重启恢复对超窗会话结转关库失败时，条目必须 +// 进入待关闭状态（经真实 NewOnlineHistoryRecorder 恢复路径验证，而非手工 +// 构造内存状态）；重连不得续接该行，清扫周期按固定边界时刻完成持久化。 +func TestRestoreStaleCloseFailureEntersPendingClose(t *testing.T) { + setupOnlineHistoryTestDB(t) + + // 预置一条超窗 Ongoing 行（dashboard 停机前遗留），时间自单一基准单调推进 + base := time.Now().Add(-3 * time.Hour).Truncate(time.Second) + staleAt := base.Add(time.Hour) + atReconnect := base.Add(2 * time.Hour) + legacy := model.OnlineSession{ServerID: 1, StartedAt: base, LastActiveAt: staleAt, Ongoing: true} + require.NoError(t, DB.Create(&legacy).Error) + + injectCloseUpdateFailure(t) + r := NewOnlineHistoryRecorder() + + stuck := getOnlineSession(t, legacy.ID) + require.True(t, stuck.Ongoing, "恢复结转关库失败时行必须保持 Ongoing 等待重试") + + // 重连：待关闭的超窗会话不得被续接，开启新会话 + r.RecordOnline(1, atReconnect) + require.Contains(t, r.open, uint64(1)) + require.NotEqual(t, legacy.ID, r.open[1].id, "待关闭会话不得被续接为活跃会话") + + dropCloseUpdateFailure(t) + r.SweepStale(atReconnect.Add(time.Second), func(uint64) (time.Time, bool) { + return atReconnect, true + }) + + closed := getOnlineSession(t, legacy.ID) + require.False(t, closed.Ongoing) + require.WithinDuration(t, staleAt, closed.EndedAt, time.Second, "旧行必须按恢复时观察到的边界闭合") + require.Equal(t, int64(2), countOnlineSessions(t, "server_id = ?", 1)) + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) +} + +func TestRecordOnlineOpensSingleSession(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + + now := time.Now() + r.RecordOnline(1, now) + r.RecordOnline(1, now.Add(time.Second)) + + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ?", 1)) + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) +} + +func TestRecordOnlinePersistsLastActiveThrottled(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + + now := time.Now() + r.RecordOnline(1, now) + // 窗口内的重复上报不应触发落库 + r.RecordOnline(1, now.Add(5*time.Second)) + s := r.open[1] + stored := getOnlineSession(t, s.id) + require.WithinDuration(t, now, stored.LastActiveAt, time.Second) + + // 超过窗口后落库 + r.RecordOnline(1, now.Add(model.OnlineActiveWindow+time.Second)) + stored = getOnlineSession(t, s.id) + require.WithinDuration(t, now.Add(model.OnlineActiveWindow+time.Second), stored.LastActiveAt, time.Second) +} + +func TestRecordOfflineClosesSession(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + + start := time.Now() + r.RecordOnline(1, start) + r.RecordOffline(1, start.Add(10*time.Minute)) + // 重复关闭应为无操作 + r.RecordOffline(1, start.Add(11*time.Minute)) + + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ?", 1)) + require.Equal(t, int64(0), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) + var s model.OnlineSession + require.NoError(t, DB.First(&s).Error) + require.WithinDuration(t, start.Add(10*time.Minute), s.EndedAt, time.Second) + require.WithinDuration(t, start, s.StartedAt, time.Second) +} + +func TestRecordOnlineContinuesSessionAfterRestart(t *testing.T) { + setupOnlineHistoryTestDB(t) + + // 模拟重启前遗留的进行中会话（最后活跃时间在在线窗口内：dashboard 只是短暂重启） + now := time.Now() + legacy := model.OnlineSession{ServerID: 1, StartedAt: now.Add(-time.Minute), LastActiveAt: now.Add(-5 * time.Second), Ongoing: true} + require.NoError(t, DB.Create(&legacy).Error) + + // 新 recorder（构造时从库恢复）：新近会话应保留并续接，不产生新会话 + r := NewOnlineHistoryRecorder() + r.RecordOnline(1, now) + + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ?", 1)) + continued := getOnlineSession(t, legacy.ID) + require.True(t, continued.Ongoing) + require.WithinDuration(t, now.Add(-time.Minute), continued.StartedAt, time.Second) +} + +func TestRestoreOngoingSessionsClosesStaleSessions(t *testing.T) { + setupOnlineHistoryTestDB(t) + + // 上次活跃时间远早于在线窗口：重启恢复时按最后已知活跃时刻结转 + staleStart := time.Now().Add(-time.Hour) + stale := model.OnlineSession{ServerID: 1, StartedAt: staleStart, LastActiveAt: staleStart.Add(30 * time.Minute), Ongoing: true} + require.NoError(t, DB.Create(&stale).Error) + + r := NewOnlineHistoryRecorder() + r.RecordOnline(1, time.Now()) + + closed := getOnlineSession(t, stale.ID) + require.False(t, closed.Ongoing) + require.WithinDuration(t, staleStart.Add(30*time.Minute), closed.EndedAt, time.Second) + // 结转后新的上报开启新会话 + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) +} + +func TestSweepStaleClosesOnlyStaleSessions(t *testing.T) { + setupOnlineHistoryTestDB(t) + r := NewOnlineHistoryRecorder() + + now := time.Now() + r.RecordOnline(1, now.Add(-time.Hour)) + // 第二台服务器：会话在 DB 但没有 servers 记录（服务器已被删除），手工放进内存 map + require.NoError(t, DB.Exec("INSERT INTO online_sessions (server_id, started_at, last_active_at, ongoing, created_at, updated_at) VALUES (2, ?, ?, 1, ?, ?)", now.Add(-time.Hour), now.Add(-time.Hour), now, now).Error) + var second model.OnlineSession + require.NoError(t, DB.Where("server_id = ?", 2).First(&second).Error) + r.mu.Lock() + r.open[2] = &onlineSessionRuntime{id: second.ID, lastPersist: now.Add(-time.Hour)} + r.mu.Unlock() + + // 服务器 1 活跃，服务器 2 不存在 → 仅关闭服务器 2 的会话 + lastActive := func(serverID uint64) (time.Time, bool) { + if serverID == 1 { + return now, true + } + return time.Time{}, false + } + r.SweepStale(now, lastActive) + + require.Equal(t, int64(1), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) + require.Equal(t, int64(0), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 2, true)) + + // 服务器 1 失联后也会被关闭，结束时刻为最后已知活跃时间 + r.SweepStale(now.Add(model.OnlineActiveWindow*2), func(uint64) (time.Time, bool) { + return now, true + }) + require.Equal(t, int64(0), countOnlineSessions(t, "server_id = ? AND ongoing = ?", 1, true)) + var closedFirst = getOnlineSession(t, func() uint64 { + var s model.OnlineSession + require.NoError(t, DB.Where("server_id = ?", 1).First(&s).Error) + return s.ID + }()) + require.WithinDuration(t, now, closedFirst.EndedAt, time.Second) +} + +func TestPruneOnlineSessions(t *testing.T) { + setupOnlineHistoryTestDB(t) + previousConf := Conf + Conf = &ConfigClass{Config: &model.Config{}} + Conf.OnlineStatsDays = 7 + t.Cleanup(func() { Conf = previousConf }) + + now := time.Now() + old := now.AddDate(0, 0, -30) + rows := []model.OnlineSession{ + {ServerID: 1, StartedAt: old, EndedAt: old.Add(time.Hour), LastActiveAt: old.Add(time.Hour)}, // 超窗口 → 清理 + {ServerID: 1, StartedAt: old, EndedAt: now.Add(-time.Hour), LastActiveAt: now.Add(-time.Hour)}, // 跨窗口边界 → 保留 + {ServerID: 1, StartedAt: now.Add(-time.Hour), EndedAt: now, LastActiveAt: now}, // 窗口内 → 保留 + {ServerID: 1, StartedAt: now, Ongoing: true, LastActiveAt: now}, // 进行中 → 保留 + {ServerID: 99, StartedAt: now.Add(-time.Hour), EndedAt: now, LastActiveAt: now}, // 服务器已删除 → 清理 + } + for i := range rows { + require.NoError(t, DB.Create(&rows[i]).Error) + } + + PruneOnlineSessions() + + require.Equal(t, int64(3), countOnlineSessions(t, "1 = 1")) + require.Equal(t, int64(1), countOnlineSessions(t, "ongoing = ?", true)) +} diff --git a/service/singleton/singleton.go b/service/singleton/singleton.go index 8d41b70b..03885020 100644 --- a/service/singleton/singleton.go +++ b/service/singleton/singleton.go @@ -63,6 +63,7 @@ func LoadSingleton(bus chan<- *model.Service) (err error) { ServerShared = NewServerClass() CronShared = NewCronClass() ServerTransferShared = NewServerTransferClass() + OnlineHistoryShared = NewOnlineHistoryRecorder() // 最后初始化 ServiceSentinel ServiceSentinelShared, err = NewServiceSentinel(bus) return @@ -94,7 +95,7 @@ func InitDBFromPath(path string) error { model.Cron{}, model.Transfer{}, model.ServerGroupServer{}, model.NAT{}, model.DDNSProfile{}, model.NotificationGroupNotification{}, model.WAF{}, model.Oauth2Bind{}, model.ServerTransfer{}, model.JWTSession{}, - model.APIToken{}, model.MCPAuditLog{}) + model.APIToken{}, model.MCPAuditLog{}, model.OnlineSession{}) if err != nil { return err }