From 010e9b90aef50292386299966900ecc108e1a736 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Sat, 10 Oct 2026 09:26:45 +0800 Subject: [PATCH] feat: record bounded heartbeat and device transition diagnostics (#378) --- server/app/goauto/device/diagnostics.go | 165 +++++++ .../goauto/device/diagnostics_mysql_test.go | 393 ++++++++++++++++ server/app/goauto/device/diagnostics_test.go | 434 ++++++++++++++++++ server/app/goauto/device/heartbeat.go | 168 +++++-- server/app/goauto/device/service.go | 26 +- server/app/goauto/migrations/migrate.go | 2 + server/app/goauto/models/agent_diagnostics.go | 33 ++ server/cmd/api/server.go | 3 + .../1791600000000_agent_diagnostics.go | 29 ++ .../1791600000000_agent_diagnostics_test.go | 48 ++ 10 files changed, 1253 insertions(+), 48 deletions(-) create mode 100644 server/app/goauto/device/diagnostics.go create mode 100644 server/app/goauto/device/diagnostics_mysql_test.go create mode 100644 server/app/goauto/device/diagnostics_test.go create mode 100644 server/app/goauto/models/agent_diagnostics.go create mode 100644 server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics.go create mode 100644 server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics_test.go diff --git a/server/app/goauto/device/diagnostics.go b/server/app/goauto/device/diagnostics.go new file mode 100644 index 0000000..d6abeed --- /dev/null +++ b/server/app/goauto/device/diagnostics.go @@ -0,0 +1,165 @@ +package device + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "time" + + "go-admin/app/goauto/models" + + log "github.com/go-admin-team/go-admin-core/logger" + "gorm.io/gorm" + "gorm.io/gorm/logger" +) + +const heartbeatHistoryBudget = 200 * time.Millisecond + +type heartbeatClientReport struct { + SnapshotSeq int64 `json:"snapshotSeq"` + FailedRounds int64 `json:"failedRounds"` + LatestFailureAt *time.Time `json:"latestFailureAt"` + LatestFailureStep *string `json:"latestFailureStep"` + LatestFailureCategory *string `json:"latestFailureCategory"` + LatestFailureDurationMs *int64 `json:"latestFailureDurationMs"` + PreviousSyncDurationMs *int64 `json:"previousSyncDurationMs"` + HasInterruptedRound bool `json:"hasInterruptedRound"` +} + +func validateHeartbeatClientReport(raw json.RawMessage) (*string, string) { + if len(raw) == 0 { + return nil, "none" + } + invalid := func() (*string, string) { return nil, "client_report_invalid" } + if len(raw) > 2048 { + return invalid() + } + var fields map[string]json.RawMessage + if json.Unmarshal(raw, &fields) != nil || len(fields) != 8 { + return invalid() + } + // encoding/json matches struct keys case-insensitively. Require every + // exact contract key before decoding so aliases cannot hide a missing key. + for _, name := range []string{"snapshotSeq", "failedRounds", "latestFailureAt", "latestFailureStep", "latestFailureCategory", "latestFailureDurationMs", "previousSyncDurationMs", "hasInterruptedRound"} { + if _, ok := fields[name]; !ok { + return invalid() + } + } + for _, name := range []string{"snapshotSeq", "failedRounds", "hasInterruptedRound"} { + if v, ok := fields[name]; !ok || bytes.Equal(bytes.TrimSpace(v), []byte("null")) { + return invalid() + } + } + var report heartbeatClientReport + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + if decoder.Decode(&report) != nil || report.SnapshotSeq < 1 || report.SnapshotSeq > 9007199254740991 || report.FailedRounds < 0 || report.FailedRounds > 1000000 { + return invalid() + } + for _, v := range []*int64{report.LatestFailureDurationMs, report.PreviousSyncDurationMs} { + if v != nil && (*v < 0 || *v > 86400000) { + return invalid() + } + } + if report.FailedRounds == 0 { + if report.LatestFailureAt != nil || report.LatestFailureStep != nil || report.LatestFailureCategory != nil || report.LatestFailureDurationMs != nil { + return invalid() + } + } else { + if report.LatestFailureAt == nil || report.LatestFailureAt.IsZero() || report.LatestFailureStep == nil || report.LatestFailureCategory == nil || report.LatestFailureDurationMs == nil { + return invalid() + } + if !oneOf(*report.LatestFailureStep, "sync", "register", "heartbeat", "recover", "flush", "claim") || !oneOf(*report.LatestFailureCategory, "api_auth", "api_conflict", "api_server", "api_other", "timeout_connect", "timeout_response", "timeout_unknown", "network", "exception") { + return invalid() + } + } + canonical, _ := json.Marshal(report) + value := string(canonical) + return &value, "accepted" +} +func oneOf(value string, allowed ...string) bool { + for _, v := range allowed { + if value == v { + return true + } + } + return false +} + +func (s *Service) saveHeartbeatHistory(ctx context.Context, row models.AgentHeartbeatLog) { + // Client cancellation after the core commit must not suppress history. There + // is deliberately no goroutine: the caller waits at most this DB deadline. + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), heartbeatHistoryBudget) + defer cancel() + err := s.DB.WithContext(writeCtx).Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent), SkipDefaultTransaction: true}).Create(&row).Error + if err != nil { + classification := "database_error" + if errors.Is(err, context.DeadlineExceeded) || writeCtx.Err() != nil { + classification = "deadline_exceeded" + } + log.Warnf("agent heartbeat history skipped: device_id=%d classification=%s", row.DeviceID, classification) + } +} + +func insertDeviceStatusEvent(tx *gorm.DB, event *models.AgentDeviceStatusEvent) error { + // Never issue a diagnostic SELECT or log the SQL error/body. + if err := tx.Session(&gorm.Session{Logger: logger.Default.LogMode(logger.Silent)}).Create(event).Error; err != nil { + return internalError(errors.New("device status event write failed")) + } + return nil +} +func logDeviceStatusEvent(event models.AgentDeviceStatusEvent) { + log.Infof("agent device status changed: device_id=%d reason=%s failed_task_count=%d order_result_unknown_count=%d", event.DeviceID, event.Reason, event.FailedTaskCount, event.OrderResultUnknownCount) +} + +const diagnosticsCleanupBatch = 5000 + +// RunDiagnosticsCleanup shares the API server lifecycle but never occupies the +// offline monitor's 15-second scan loop. +func RunDiagnosticsCleanup(ctx context.Context, s *Service, onError func(error)) { + heartbeats := time.NewTicker(time.Hour) + defer heartbeats.Stop() + events := time.NewTicker(24 * time.Hour) + defer events.Stop() + runDiagnosticsCleanup(ctx, s, heartbeats.C, events.C, onError) +} +func runDiagnosticsCleanup(ctx context.Context, s *Service, heartbeats, events <-chan time.Time, onError func(error)) { + for { + select { + case <-ctx.Done(): + return + case <-heartbeats: + if _, err := s.CleanupHeartbeatHistory(ctx); err != nil && ctx.Err() == nil && onError != nil { + onError(err) + } + case <-events: + if _, err := s.CleanupDeviceStatusEvents(ctx); err != nil && ctx.Err() == nil && onError != nil { + onError(err) + } + } + } +} + +func (s *Service) CleanupHeartbeatHistory(ctx context.Context) (int64, error) { + return cleanupDiagnostics(s.DB.WithContext(ctx), &models.AgentHeartbeatLog{}, "received_at", s.Now().Add(-7*24*time.Hour)) +} +func (s *Service) CleanupDeviceStatusEvents(ctx context.Context) (int64, error) { + return cleanupDiagnostics(s.DB.WithContext(ctx), &models.AgentDeviceStatusEvent{}, "occurred_at", s.Now().Add(-30*24*time.Hour)) +} +func cleanupDiagnostics(db *gorm.DB, model any, timestamp string, cutoff time.Time) (int64, error) { + // A fixed-size ID batch works on both MySQL and SQLite. Recheck expiration + // in DELETE, and never loop through an unbounded backlog in one run. + var ids []uint64 + if err := db.Model(model).Where(timestamp+" < ?", cutoff).Order(timestamp+", id").Limit(diagnosticsCleanupBatch).Pluck("id", &ids).Error; err != nil { + return 0, errors.New("diagnostics cleanup read failed") + } + if len(ids) == 0 { + return 0, nil + } + result := db.Where("id IN ? AND "+timestamp+" < ?", ids, cutoff).Delete(model) + if result.Error != nil { + return 0, errors.New("diagnostics cleanup delete failed") + } + return result.RowsAffected, nil +} diff --git a/server/app/goauto/device/diagnostics_mysql_test.go b/server/app/goauto/device/diagnostics_mysql_test.go new file mode 100644 index 0000000..5e78676 --- /dev/null +++ b/server/app/goauto/device/diagnostics_mysql_test.go @@ -0,0 +1,393 @@ +package device_test + +import ( + "context" + "errors" + "net" + "os" + "strings" + "sync" + "testing" + "time" + + . "go-admin/app/goauto/device" + "go-admin/app/goauto/models" + versionlocal "go-admin/cmd/migrate/migration/version-local" + common "go-admin/common/models" + + driver "github.com/go-sql-driver/mysql" + "github.com/google/uuid" + "gorm.io/driver/mysql" + "gorm.io/gorm" + "gorm.io/gorm/clause" + "gorm.io/gorm/logger" +) + +// Explicit opt-in only; refuses production/remote DSNs and any nonempty schema. +// Credentials are injected by the caller and never included in test errors. +func TestMySQLAgentDiagnostics(t *testing.T) { + raw := os.Getenv("GOAUTO_378_TEST_MYSQL_DSN") + if raw == "" { + t.Skip("isolated local MySQL DSN not provided") + } + cfg, err := driver.ParseDSN(raw) + if err != nil { + t.Fatal("invalid test DSN") + } + host, _, err := net.SplitHostPort(cfg.Addr) + if err != nil || cfg.Net != "tcp" || (host != "127.0.0.1" && host != "localhost" && host != "::1") || cfg.DBName != "goauto_378_test" { + t.Fatal("requires local tcp database goauto_378_test") + } + cfg.ParseTime = true + db, err := gorm.Open(mysql.Open(cfg.FormatDSN()), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)}) + if err != nil { + t.Fatal("test MySQL connection failed") + } + sqlDB, err := db.DB() + if err != nil { + t.Fatal("test database unavailable") + } + t.Cleanup(func() { _ = sqlDB.Close() }) + sqlDB.SetMaxOpenConns(8) + var version string + if db.Raw("SELECT VERSION()").Scan(&version).Error != nil || !strings.HasPrefix(version, "8.") { + t.Fatal("requires MySQL 8") + } + tables, err := db.Migrator().GetTables() + if err != nil || len(tables) != 0 { + t.Fatal("test schema must be empty") + } + owned := []any{&common.Migration{}, &models.AgentDevice{}, &models.PDDProduct{}, &models.CollectionRule{}, &models.CollectionTask{}, &models.PurchaseTask{}, &models.PurchaseTaskAttempt{}, &models.PDDAccount{}, &models.ShopeeProduct{}, &models.SYBProduct{}} + t.Cleanup(func() { + for _, model := range append([]any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}}, owned...) { + if err := db.Migrator().DropTable(model); err != nil { + t.Errorf("test table cleanup failed for %T", model) + } + } + }) + if err := db.AutoMigrate(owned...); err != nil { + t.Fatal("test base migration failed") + } + const migrationVersion = "1791600000000_agent_diagnostics" + for i := 0; i < 2; i++ { + if versionlocal.MigrateAgentDiagnostics(db, migrationVersion) != nil { + t.Fatal("diagnostic migration failed") + } + } + for _, idx := range []struct { + model any + name string + }{{&models.AgentHeartbeatLog{}, "ix_heartbeat_device_received"}, {&models.AgentHeartbeatLog{}, "ix_heartbeat_received"}, {&models.AgentDeviceStatusEvent{}, "ix_device_event_occurred"}} { + if !db.Migrator().HasIndex(idx.model, idx.name) { + t.Fatalf("missing MySQL index %s", idx.name) + } + } + // Exercise reviewed rollback only on fresh diagnostic tables in this schema. + if db.Migrator().DropTable(&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}) != nil { + t.Fatal("diagnostic test rollback failed") + } + if db.Where("version = ?", migrationVersion).Delete(&common.Migration{}).Error != nil { + t.Fatal("test version rollback failed") + } + if versionlocal.MigrateAgentDiagnostics(db, migrationVersion) != nil { + t.Fatal("diagnostic reapply failed") + } + seed := func(t *testing.T) (*Service, RegisterResponse, string) { + t.Helper() + s := NewService(db) + s.Now = func() time.Time { return time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC) } + token := "isolated-mysql-" + uuid.NewString() + s.GenerateToken = func() (string, error) { return token, nil } + d, err := s.Register(context.Background(), RegisterRequest{RequestID: uuid.NewString(), InstallID: uuid.NewString(), Name: "test", Manufacturer: "test", Model: "test", AndroidVersion: "14", AgentVersion: "test", PDDVersion: "test"}, "") + if err != nil { + t.Fatal("synthetic device registration failed") + } + if db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", s.Now().Add(-time.Minute)).Error != nil { + t.Fatal("test seed failed") + } + t.Cleanup(func() { + if db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("status", models.DeviceStatusDisabled).Error != nil { + t.Error("synthetic device cleanup failed") + } + }) + return s, d, token + } + + t.Run("scan_revalidates_heartbeat_after_lock_wait", func(t *testing.T) { + s, d, _ := seed(t) + lock := db.Begin() + if lock.Error != nil { + t.Fatal("begin failed") + } + defer lock.Rollback() + var device models.AgentDevice + if lock.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, d.DeviceID).Error != nil { + t.Fatal("lock failed") + } + done := make(chan struct { + n int64 + err error + }, 1) + go func() { + n, e := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + done <- struct { + n int64 + err error + }{n, e} + }() + select { + case <-done: + t.Fatal("scan did not wait for device lock") + case <-time.After(100 * time.Millisecond): + } + if lock.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", s.Now()).Error != nil { + t.Fatal("refresh failed") + } + if lock.Commit().Error != nil { + t.Fatal("commit failed") + } + select { + case result := <-done: + if result.err != nil || result.n != 0 { + t.Fatalf("fresh device falsely marked offline: count=%d error=%v", result.n, result.err != nil) + } + case <-time.After(5 * time.Second): + t.Fatal("scan stuck") + } + }) + t.Run("concurrent_scans_and_resume_are_unique", func(t *testing.T) { + s, d, testDeviceToken := seed(t) + var wg sync.WaitGroup + errs := make(chan error, 2) + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + errs <- err + }() + } + wg.Wait() + for i := 0; i < 2; i++ { + if err := <-errs; err != nil { + var dbErr *driver.MySQLError + if errors.As(err, &dbErr) { + t.Fatalf("concurrent scan failed mysql_errno=%d", dbErr.Number) + } + t.Fatalf("concurrent scan failed classification=%T", err) + } + } + req := HeartbeatRequest{RequestID: uuid.NewString()} + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, err := s.Heartbeat(context.Background(), req, testDeviceToken) + errs <- err + }() + } + wg.Wait() + for i := 0; i < 2; i++ { + if <-errs != nil { + t.Fatal("concurrent heartbeat failed") + } + } + var events []models.AgentDeviceStatusEvent + db.Where("device_id = ?", d.DeviceID).Order("id").Find(&events) + if len(events) != 2 || events[0].Reason != "heartbeat_timeout" || events[1].Reason != "heartbeat_resumed" { + t.Fatalf("transition count=%d", len(events)) + } + var n int64 + db.Model(&models.AgentHeartbeatLog{}).Where("device_id = ?", d.DeviceID).Count(&n) + if n != 1 { + t.Fatalf("replay history count=%d", n) + } + }) + t.Run("event_failure_rolls_back_real_mysql", func(t *testing.T) { + s, d, _ := seed(t) + product := models.PDDProduct{GoodsID: "37801", URL: "https://example.invalid/test"} + rule := models.CollectionRule{Name: "test", ContentJSON: `{}`} + if db.Create(&product).Error != nil || db.Create(&rule).Error != nil { + t.Fatal("synthetic task input failed") + } + task := models.CollectionTask{PDDProductID: &product.ID, RuleID: rule.ID, DeviceID: &d.DeviceID, Status: models.TaskStatusRunning, URLSnapshot: product.URL, GoodsIDSnapshot: product.GoodsID, RuleSnapshot: rule.ContentJSON} + if db.Create(&task).Error != nil { + t.Fatal("synthetic task creation failed") + } + if db.Exec("CREATE TRIGGER diagnostic_test_reject BEFORE INSERT ON agent_device_status_event FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'synthetic_failure'").Error != nil { + t.Fatal("test trigger creation failed") + } + defer db.Exec("DROP TRIGGER IF EXISTS diagnostic_test_reject") + n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + if err == nil || n != 0 { + t.Fatal("failed event did not abort") + } + var device models.AgentDevice + db.First(&device, d.DeviceID) + if device.Status != models.DeviceStatusOnline { + t.Fatal("state escaped rollback") + } + var storedTask models.CollectionTask + if db.First(&storedTask, task.ID).Error != nil || storedTask.Status != models.TaskStatusRunning || storedTask.DeviceRunSlot == nil { + t.Fatal("task escaped rollback") + } + if db.Exec("DROP TRIGGER diagnostic_test_reject").Error != nil { + t.Fatal("test trigger removal failed") + } + n, err = s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + if err != nil || n != 1 { + t.Fatal("retry failed") + } + var event models.AgentDeviceStatusEvent + if db.Where("device_id = ?", d.DeviceID).First(&event).Error != nil || event.FailedTaskCount != 1 || event.OrderResultUnknownCount != 0 { + t.Fatal("per-device counts incorrect") + } + }) + for _, domain := range []string{"collection", "purchase"} { + t.Run("scan_does_not_deadlock_task_then_device_reset_"+domain, func(t *testing.T) { + s, d, _ := seed(t) + goodsID := "37802" + if domain == "purchase" { + goodsID = "37803" + } + product := models.PDDProduct{GoodsID: goodsID, URL: "https://example.invalid/test"} + rule := models.CollectionRule{Name: "test", ContentJSON: `{}`} + if db.Create(&product).Error != nil || db.Create(&rule).Error != nil { + t.Fatal("synthetic inputs failed") + } + var taskID uint64 + var taskModel any + if domain == "collection" { + task := models.CollectionTask{PDDProductID: &product.ID, RuleID: rule.ID, DeviceID: &d.DeviceID, Status: models.TaskStatusRunning, URLSnapshot: product.URL, GoodsIDSnapshot: product.GoodsID, RuleSnapshot: rule.ContentJSON} + if db.Create(&task).Error != nil { + t.Fatal("synthetic task failed") + } + taskID = task.ID + taskModel = &models.CollectionTask{} + } else { + task := models.PurchaseTask{PDDProductID: product.ID, DeviceID: &d.DeviceID, ExecutionMode: models.PurchaseExecutionModeRehearsal, Status: models.PurchaseTaskStatusRunning, PDDURLSnapshot: product.URL, PDDGoodsIDSnapshot: product.GoodsID, Quantity: 1, ReferenceUnitPriceCent: 100, MinUnitPriceCent: 20, MaxUnitPriceCent: 150, Currency: "CNY", RuleType: "pddPurchase", RuleSchemaVersion: 1, RequiredCapabilitiesJSON: `[]`, RuleSnapshot: `{}`, CreateRequestID: uuid.NewString()} + if db.Create(&task).Error != nil { + t.Fatal("synthetic purchase task failed") + } + taskID = task.ID + taskModel = &models.PurchaseTask{} + } + resetTx := db.Begin() + defer resetTx.Rollback() + if resetTx.Clauses(clause.Locking{Strength: "UPDATE"}).First(taskModel, taskID).Error != nil { + t.Fatal("reset task lock failed") + } + type scanMarker struct{} + deviceLocked, releaseScan := make(chan struct{}), make(chan struct{}) + callback := "diagnostic_test_scan_lock_barrier" + if db.Callback().Query().After("gorm:query").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Context.Value(scanMarker{}) == true && tx.Statement.Table == "agent_device" { + if _, ok := tx.Statement.Clauses["FOR"]; ok { + close(deviceLocked) + <-releaseScan + } + } + }) != nil { + t.Fatal("register scan barrier failed") + } + defer db.Callback().Query().Remove(callback) + scanDone, resetDone := make(chan error, 1), make(chan error, 1) + go func() { + _, err := s.MarkStaleDevicesOffline(context.WithValue(context.Background(), scanMarker{}, true), DefaultOfflineThreshold) + scanDone <- err + }() + select { + case <-deviceLocked: + case <-time.After(5 * time.Second): + close(releaseScan) + t.Fatal("scan device lock not reached") + } + go func() { + var device models.AgentDevice + err := resetTx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, d.DeviceID).Error + resetTx.Rollback() + resetDone <- err + }() + close(releaseScan) + scanErr, resetErr := <-scanDone, <-resetDone + if resetErr != nil { + var mysqlErr *driver.MySQLError + if errors.As(resetErr, &mysqlErr) { + t.Fatalf("reset lock failed mysql_errno=%d", mysqlErr.Number) + } + t.Fatal("reset lock failed") + } + var mysqlErr *driver.MySQLError + if scanErr == nil || !errors.As(scanErr, &mysqlErr) || mysqlErr.Number != 3572 { + if mysqlErr != nil { + t.Fatalf("scan must decline busy task without deadlock; mysql_errno=%d", mysqlErr.Number) + } + t.Fatal("scan must decline busy task") + } + var eventCount int64 + db.Model(&models.AgentDeviceStatusEvent{}).Where("device_id = ?", d.DeviceID).Count(&eventCount) + if eventCount != 0 { + t.Fatal("busy task created event") + } + var device models.AgentDevice + db.First(&device, d.DeviceID) + if device.Status != models.DeviceStatusOnline { + t.Fatal("busy task changed status") + } + db.Callback().Query().Remove(callback) + n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + if err != nil || n != 1 { + t.Fatal("scan did not retry after reset released locks") + } + }) + } + t.Run("history_timeout_is_bounded_on_real_lock", func(t *testing.T) { + s, _, testDeviceToken := seed(t) + // Hold the diagnostic table's insert gap; liveness updates remain free. + lock := db.Begin() + defer lock.Rollback() + if lock.Exec("SELECT id FROM agent_heartbeat_log WHERE id > 0 FOR UPDATE").Error != nil { + t.Fatal("history lock failed") + } + started := time.Now() + res, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken) + t.Logf("heartbeat with locked history table elapsed=%s", time.Since(started)) + if err != nil || !res.Online || time.Since(started) > time.Second { + t.Fatal("history lock changed heartbeat outcome or exceeded budget") + } + }) + t.Run("strict_cleanup", func(t *testing.T) { + s, d, _ := seed(t) + boundary := models.AgentHeartbeatLog{DeviceID: d.DeviceID, ReceivedAt: s.Now().Add(-7 * 24 * time.Hour), RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"} + expired := boundary + expired.ReceivedAt = expired.ReceivedAt.Add(-time.Second) + db.Create(&boundary) + db.Create(&expired) + n, err := s.CleanupHeartbeatHistory(context.Background()) + if err != nil || n != 1 { + t.Fatalf("cleanup=%d failed=%v", n, err != nil) + } + history := make([]models.AgentHeartbeatLog, 5001) + events := make([]models.AgentDeviceStatusEvent, 5001) + for i := range history { + history[i] = models.AgentHeartbeatLog{DeviceID: d.DeviceID, ReceivedAt: s.Now().Add(-31 * 24 * time.Hour), ClientReportStatus: "none"} + events[i] = models.AgentDeviceStatusEvent{DeviceID: d.DeviceID, OccurredAt: s.Now().Add(-31 * 24 * time.Hour)} + } + if db.CreateInBatches(&history, 500).Error != nil || db.CreateInBatches(&events, 500).Error != nil { + t.Fatal("synthetic cleanup batch seed failed") + } + n, err = s.CleanupHeartbeatHistory(context.Background()) + if err != nil || n != 5000 { + t.Fatalf("history cleanup batch=%d failed=%v", n, err != nil) + } + n, err = s.CleanupDeviceStatusEvents(context.Background()) + if err != nil || n != 5000 { + t.Fatalf("event cleanup batch=%d failed=%v", n, err != nil) + } + var remaining int64 + db.Model(&models.AgentHeartbeatLog{}).Where("device_id = ?", d.DeviceID).Count(&remaining) + if remaining != 2 { + t.Fatalf("history boundary/remaining=%d", remaining) + } + }) +} diff --git a/server/app/goauto/device/diagnostics_test.go b/server/app/goauto/device/diagnostics_test.go new file mode 100644 index 0000000..f6a2f7b --- /dev/null +++ b/server/app/goauto/device/diagnostics_test.go @@ -0,0 +1,434 @@ +package device + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "go-admin/app/goauto/models" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "gorm.io/gorm" +) + +func TestHeartbeatDiagnosticsHistoryAndReportIsolation(t *testing.T) { + db := openTestDatabase(t) + service := newTestService(t, db) + _, registered := registerHeartbeatDevice(t, service) + gin.SetMode(gin.TestMode) + router := gin.New() + router.POST("/heartbeat", Handler{DB: db}.Heartbeat) + for _, tc := range []struct{ name, report, status string }{ + {"missing", "", "none"}, + {"accepted", `,"clientReport":{"snapshotSeq":1,"failedRounds":0,"latestFailureAt":null,"latestFailureStep":null,"latestFailureCategory":null,"latestFailureDurationMs":null,"previousSyncDurationMs":25,"hasInterruptedRound":false}`, "accepted"}, + {"wrong_type", `,"clientReport":"secret-not-stored"`, "client_report_invalid"}, + {"unknown_field", `,"clientReport":{"url":"secret-not-stored"}`, "client_report_invalid"}, + } { + t.Run(tc.name, func(t *testing.T) { + id := uuid.NewString() + body := fmt.Sprintf(`{"requestId":%q,"currentTaskId":null%s}`, id, tc.report) + for i := 0; i < 2; i++ { + req := httptest.NewRequest(http.MethodPost, "/heartbeat", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+testDeviceToken) + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + if res.Code != 200 { + t.Fatalf("heartbeat status=%d body=%s", res.Code, res.Body.String()) + } + var envelope struct { + Data map[string]any `json:"data"` + } + _ = json.Unmarshal(res.Body.Bytes(), &envelope) + if envelope.Data["acceptsClientReport"] != true { + t.Fatal("capability missing") + } + } + var rows []struct { + DeviceID uint64 + ClientReportStatus string + ClientReportJSON *string + } + if err := db.Table("agent_heartbeat_log").Where("request_id = ?", id).Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].DeviceID != registered.DeviceID || rows[0].ClientReportStatus != tc.status || (rows[0].ClientReportJSON != nil) != (tc.status == "accepted") { + t.Fatalf("history=%+v", rows) + } + }) + } +} + +func TestDisableWritesOneAtomicStatusEvent(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + _, d := registerHeartbeatDevice(t, s) + for i := 0; i < 2; i++ { + if err := s.Disable(context.Background(), d.DeviceID); err != nil { + t.Fatal(err) + } + } + var events []models.AgentDeviceStatusEvent + if err := db.Find(&events).Error; err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].Reason != "disabled" || events[0].FromStatus != "online" || events[0].ToStatus != "disabled" { + t.Fatalf("events=%+v", events) + } +} + +func TestStatusEventFailureRollsBackAndCanRetry(t *testing.T) { + for _, action := range []string{"timeout", "resumed", "disabled"} { + t.Run(action, func(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + _, d := registerHeartbeatDevice(t, s) + status := models.DeviceStatusOnline + if action == "resumed" { + status = models.DeviceStatusOffline + } + old := s.Now().Add(-time.Minute) + if err := db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Updates(map[string]any{"status": status, "last_heartbeat_at": old}).Error; err != nil { + t.Fatal(err) + } + var task models.CollectionTask + if action == "timeout" || action == "disabled" { + product := models.PDDProduct{GoodsID: "1001", URL: "https://example.invalid/test"} + rule := models.CollectionRule{Name: "test", ContentJSON: `{}`} + if err := db.Create(&product).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&rule).Error; err != nil { + t.Fatal(err) + } + task = newHeartbeatTask(product, rule, d.DeviceID) + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + } + operation := func() error { + switch action { + case "timeout": + _, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + return err + case "disabled": + return s.Disable(context.Background(), d.DeviceID) + default: + _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken) + return err + } + } + callback := "diagnostic_event_failure" + if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Table == "agent_device_status_event" { + tx.AddError(errors.New("injected private error")) + } + }); err != nil { + t.Fatal(err) + } + if err := operation(); err == nil { + t.Fatal("event failure must abort transaction") + } + var stored models.AgentDevice + if err := db.First(&stored, d.DeviceID).Error; err != nil { + t.Fatal(err) + } + if stored.Status != status || !stored.LastHeartbeatAt.Equal(old) { + t.Fatalf("state escaped rollback=%+v", stored.Status) + } + if task.ID != 0 { + var storedTask models.CollectionTask + if err := db.First(&storedTask, task.ID).Error; err != nil { + t.Fatal(err) + } + if storedTask.Status != models.TaskStatusRunning || storedTask.DeviceRunSlot == nil { + t.Fatal("task changes escaped rollback") + } + } + _ = db.Callback().Create().Remove(callback) + if err := operation(); err != nil { + t.Fatal(err) + } + var n int64 + db.Model(&models.AgentDeviceStatusEvent{}).Count(&n) + if n != 1 { + t.Fatalf("events=%d", n) + } + }) + } +} + +func TestOfflineEventCountsPerDeviceIncludingUnknownBoundary(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + ids := make([]uint64, 3) + for i := range ids { + s.GenerateToken = func() (string, error) { return fmt.Sprintf("test-token-%d", i), nil } + _, registered := registerHeartbeatDevice(t, s) + ids[i] = registered.DeviceID + if err := db.Model(&models.AgentDevice{}).Where("id = ?", ids[i]).Update("last_heartbeat_at", s.Now().Add(-time.Minute)).Error; err != nil { + t.Fatal(err) + } + } + product := models.PDDProduct{GoodsID: "1002", URL: "https://example.invalid/test"} + rule := models.CollectionRule{Name: "test", ContentJSON: `{}`} + if err := db.Create(&product).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&rule).Error; err != nil { + t.Fatal(err) + } + collection := newHeartbeatTask(product, rule, ids[0]) + if err := db.Create(&collection).Error; err != nil { + t.Fatal(err) + } + for i, status := range []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted} { + task := newHeartbeatPurchaseTask(product, ids[i], status) + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + attempt := newRunningPurchaseAttempt(task.ID, ids[i]) + if err := db.Create(&attempt).Error; err != nil { + t.Fatal(err) + } + } + if n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold); err != nil || n != 3 { + t.Fatalf("scan=%d %v", n, err) + } + var events []models.AgentDeviceStatusEvent + if err := db.Order("device_id").Find(&events).Error; err != nil { + t.Fatal(err) + } + if len(events) != 3 || events[0].FailedTaskCount != 2 || events[0].OrderResultUnknownCount != 0 || events[1].FailedTaskCount != 0 || events[1].OrderResultUnknownCount != 1 || events[2].FailedTaskCount != 0 || events[2].OrderResultUnknownCount != 0 { + t.Fatalf("events=%+v", events) + } +} + +func TestHeartbeatHistoryOnlyForSuccessfulMainRequest(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + registerHeartbeatDevice(t, s) + if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, "wrong-token"); err == nil { + t.Fatal("bad token accepted") + } + if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: "invalid"}, testDeviceToken); err == nil { + t.Fatal("bad request accepted") + } + id := uint64(999) + if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString(), CurrentTaskID: &id}, testDeviceToken); err == nil { + t.Fatal("task mismatch accepted") + } + var count int64 + if err := db.Model(&models.AgentHeartbeatLog{}).Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatal("failed request stored history") + } +} + +func TestHeartbeatHistoryBoundedIndependentAndBestEffort(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + _, d := registerHeartbeatDevice(t, s) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + callback := "history_wait_for_deadline" + if err := db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) { + if tx.Statement.Table != "agent_heartbeat_log" { + return + } + // We are already outside the core transaction; cancellation must not + // cancel the independent history context. + cancel() + if tx.Statement.Context.Err() != nil { + t.Error("history inherited request cancellation") + } + deadline, ok := tx.Statement.Context.Deadline() + if !ok || time.Until(deadline) > 201*time.Millisecond { + t.Error("missing bounded deadline") + } + var current models.AgentDevice + if err := db.First(¤t, d.DeviceID).Error; err != nil || current.LastHeartbeatAt == nil { + t.Error("core was not committed before history") + } + <-tx.Statement.Context.Done() + tx.AddError(tx.Statement.Context.Err()) + }); err != nil { + t.Fatal(err) + } + started := time.Now() + res, err := s.Heartbeat(ctx, HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken) + if err != nil || !res.Online || time.Since(started) > time.Second { + t.Fatalf("best effort changed result: %v", err) + } + _ = db.Callback().Create().Remove(callback) + if err := db.Migrator().DropTable(&models.AgentHeartbeatLog{}); err != nil { + t.Fatal(err) + } + if _, err := s.Heartbeat(context.Background(), HeartbeatRequest{RequestID: uuid.NewString()}, testDeviceToken); err != nil { + t.Fatal(err) + } +} + +func TestClientReportWhitelistBounds(t *testing.T) { + valid := `{"snapshotSeq":1,"failedRounds":1,"latestFailureAt":"2026-10-10T00:00:00Z","latestFailureStep":"heartbeat","latestFailureCategory":"network","latestFailureDurationMs":12,"previousSyncDurationMs":null,"hasInterruptedRound":false}` + for _, tc := range []struct{ name, raw, status string }{ + {"accepted", valid, "accepted"}, {"none", "", "none"}, {"null", "null", "client_report_invalid"}, {"array", "[]", "client_report_invalid"}, + {"sequence", strings.Replace(valid, `"snapshotSeq":1`, `"snapshotSeq":0`, 1), "client_report_invalid"}, + {"duration", strings.Replace(valid, `"latestFailureDurationMs":12`, `"latestFailureDurationMs":86400001`, 1), "client_report_invalid"}, + {"unknown_step", strings.Replace(valid, `"heartbeat"`, `"http://private"`, 1), "client_report_invalid"}, + {"inconsistent", strings.Replace(valid, `"failedRounds":1`, `"failedRounds":0`, 1), "client_report_invalid"}, + {"unknown", strings.Replace(valid, `"snapshotSeq"`, `"privateText"`, 1), "client_report_invalid"}, + {"null_bool", strings.Replace(valid, `"hasInterruptedRound":false`, `"hasInterruptedRound":null`, 1), "client_report_invalid"}, + {"case_alias_cannot_replace_missing_required", `{"snapshotSeq":1,"failedRounds":0,"latestFailureAt":null,"latestFailureStep":null,"latestFailureCategory":null,"SNAPSHOTSEQ":2,"previousSyncDurationMs":null,"hasInterruptedRound":false}`, "client_report_invalid"}, + {"oversize", strings.Repeat(" ", 2049) + valid, "client_report_invalid"}, + } { + t.Run(tc.name, func(t *testing.T) { + value, status := validateHeartbeatClientReport(json.RawMessage(tc.raw)) + if status != tc.status || (status != "accepted" && value != nil) { + t.Fatalf("status=%s valuePresent=%v", status, value != nil) + } + }) + } +} + +func TestDiagnosticsCleanupIsBoundedAndStrictlyExpired(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + now := s.Now() + old := now.Add(-31 * 24 * time.Hour) + heartbeats := make([]models.AgentHeartbeatLog, 5001) + events := make([]models.AgentDeviceStatusEvent, 5001) + for i := range heartbeats { + heartbeats[i] = models.AgentHeartbeatLog{DeviceID: 1, ReceivedAt: old, RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"} + events[i] = models.AgentDeviceStatusEvent{DeviceID: 1, OccurredAt: old, FromStatus: "online", ToStatus: "offline", Reason: "heartbeat_timeout"} + } + if err := db.CreateInBatches(&heartbeats, 100).Error; err != nil { + t.Fatal(err) + } + if err := db.CreateInBatches(&events, 100).Error; err != nil { + t.Fatal(err) + } + boundaryHeartbeat := models.AgentHeartbeatLog{DeviceID: 1, ReceivedAt: now.Add(-7 * 24 * time.Hour), RequestID: uuid.NewString(), AgentVersion: "test", ClientReportStatus: "none"} + boundaryEvent := models.AgentDeviceStatusEvent{DeviceID: 1, OccurredAt: now.Add(-30 * 24 * time.Hour), FromStatus: "online", ToStatus: "offline", Reason: "heartbeat_timeout"} + db.Create(&boundaryHeartbeat) + db.Create(&boundaryEvent) + n, err := s.CleanupHeartbeatHistory(context.Background()) + if err != nil || n != 5000 { + t.Fatalf("heartbeat cleanup=%d %v", n, err) + } + n, err = s.CleanupDeviceStatusEvents(context.Background()) + if err != nil || n != 5000 { + t.Fatalf("event cleanup=%d %v", n, err) + } + var count int64 + db.Model(&models.AgentHeartbeatLog{}).Count(&count) + if count != 2 { + t.Fatalf("remaining=%d", count) + } + db.Model(&models.AgentDeviceStatusEvent{}).Count(&count) + if count != 2 { + t.Fatalf("remaining=%d", count) + } + n, err = s.CleanupHeartbeatHistory(context.Background()) + if err != nil || n != 1 { + t.Fatalf("strict heartbeat cleanup=%d %v", n, err) + } + n, err = s.CleanupDeviceStatusEvents(context.Background()) + if err != nil || n != 1 { + t.Fatalf("strict event cleanup=%d %v", n, err) + } +} + +func TestDiagnosticsCleanupRunsOnSeparateSchedulesAndStops(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + old := s.Now().Add(-31 * 24 * time.Hour) + history := models.AgentHeartbeatLog{ReceivedAt: old, ClientReportStatus: "none"} + event := models.AgentDeviceStatusEvent{OccurredAt: old} + if err := db.Create(&history).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&event).Error; err != nil { + t.Fatal(err) + } + heartbeats, events := make(chan time.Time), make(chan time.Time) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + runDiagnosticsCleanup(ctx, s, heartbeats, events, func(err error) { t.Error(err) }) + }() + heartbeats <- s.Now() + // A second tick synchronizes completion of the first cleanup. + heartbeats <- s.Now() + var n int64 + db.Model(&models.AgentHeartbeatLog{}).Count(&n) + if n != 0 { + t.Fatal("hourly history cleanup did not run") + } + db.Model(&models.AgentDeviceStatusEvent{}).Count(&n) + if n != 1 { + t.Fatal("hourly tick cleaned daily events") + } + events <- s.Now() + events <- s.Now() + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("cleanup did not stop") + } + db.Model(&models.AgentDeviceStatusEvent{}).Count(&n) + if n != 0 { + t.Fatal("daily event cleanup did not run") + } +} + +func TestOfflineDiagnosticsAtomicAndResumed(t *testing.T) { + db := openTestDatabase(t) + s := newTestService(t, db) + _, d := registerHeartbeatDevice(t, s) + old := s.Now().Add(-time.Minute) + if err := db.Model(&models.AgentDevice{}).Where("id = ?", d.DeviceID).Update("last_heartbeat_at", old).Error; err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + n, err := s.MarkStaleDevicesOffline(context.Background(), DefaultOfflineThreshold) + if err != nil || n != int64(1-i) { + t.Fatalf("scan=%d %v", n, err) + } + } + var events []struct { + Reason, FromStatus, ToStatus string + LastHeartbeatAt *time.Time + FailedTaskCount, OrderResultUnknownCount int64 + } + if err := db.Table("agent_device_status_event").Find(&events).Error; err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].Reason != "heartbeat_timeout" || events[0].FromStatus != "online" || events[0].ToStatus != "offline" || events[0].LastHeartbeatAt == nil || !events[0].LastHeartbeatAt.Equal(old) { + t.Fatalf("events=%+v", events) + } + req := HeartbeatRequest{RequestID: uuid.NewString()} + for i := 0; i < 2; i++ { + if _, err := s.Heartbeat(context.Background(), req, testDeviceToken); err != nil { + t.Fatal(err) + } + } + events = nil + if err := db.Table("agent_device_status_event").Order("id").Find(&events).Error; err != nil { + t.Fatal(err) + } + if len(events) != 2 || events[1].Reason != "heartbeat_resumed" { + t.Fatalf("events=%+v", events) + } +} diff --git a/server/app/goauto/device/heartbeat.go b/server/app/goauto/device/heartbeat.go index f0138d2..14b92c8 100644 --- a/server/app/goauto/device/heartbeat.go +++ b/server/app/goauto/device/heartbeat.go @@ -2,6 +2,7 @@ package device import ( "context" + "encoding/json" "errors" "fmt" "time" @@ -10,6 +11,7 @@ import ( "github.com/google/uuid" "gorm.io/gorm" + "gorm.io/gorm/clause" ) const ( @@ -18,9 +20,10 @@ const ( ) type HeartbeatRequest struct { - RequestID string `json:"requestId"` - CurrentTaskID *uint64 `json:"currentTaskId"` - Capabilities []string `json:"capabilities,omitempty"` + RequestID string `json:"requestId"` + CurrentTaskID *uint64 `json:"currentTaskId"` + Capabilities []string `json:"capabilities,omitempty"` + ClientReport json.RawMessage `json:"clientReport,omitempty"` } type HeartbeatResponse struct { @@ -32,6 +35,7 @@ type HeartbeatResponse struct { ServerTime string `json:"serverTime"` HeartbeatIntervalSeconds int `json:"heartbeatIntervalSeconds"` Replayed bool `json:"replayed,omitempty"` + AcceptsClientReport bool `json:"acceptsClientReport"` } func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest, presentedToken string) (HeartbeatResponse, error) { @@ -50,10 +54,12 @@ func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest, interval = DefaultHeartbeatIntervalSeconds } var response HeartbeatResponse + var history models.AgentHeartbeatLog + var resumed *models.AgentDeviceStatusEvent err = service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var device models.AgentDevice digest := tokenDigest(presentedToken) - if err := tx.Where("token_digest = ? AND token_revoked_at IS NULL", digest).First(&device).Error; err != nil { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("token_digest = ? AND token_revoked_at IS NULL", digest).First(&device).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return tokenInvalidError() } @@ -93,14 +99,30 @@ func (service *Service) Heartbeat(ctx context.Context, request HeartbeatRequest, if result.RowsAffected == 0 { return tokenInvalidError() } + if device.Status == models.DeviceStatusOffline { + resumed = &models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusOnline, Reason: "heartbeat_resumed", LastHeartbeatAt: device.LastHeartbeatAt} + if err := insertDeviceStatusEvent(tx, resumed); err != nil { + return err + } + } } response = HeartbeatResponse{ RequestID: request.RequestID, DeviceID: device.ID, CurrentTaskID: runningTaskID, Online: true, Busy: runningTaskID != nil, ServerTime: now.UTC().Format(time.RFC3339Nano), - HeartbeatIntervalSeconds: interval, Replayed: replayed, + HeartbeatIntervalSeconds: interval, Replayed: replayed, AcceptsClientReport: true, } + history = models.AgentHeartbeatLog{DeviceID: device.ID, ReceivedAt: now, RequestID: request.RequestID, CurrentTaskID: request.CurrentTaskID, RunningTaskID: runningTaskID, Busy: runningTaskID != nil, AgentVersion: device.AgentVersion} return nil }) + if err == nil { + if resumed != nil { + logDeviceStatusEvent(*resumed) + } + if !response.Replayed { + history.ClientReportJSON, history.ClientReportStatus = validateHeartbeatClientReport(request.ClientReport) + service.saveHeartbeatHistory(ctx, history) + } + } return response, err } @@ -124,50 +146,107 @@ func (service *Service) MarkStaleDevicesOffline(ctx context.Context, threshold t now := service.Now() cutoff := now.Add(-threshold) var changed int64 + var events []models.AgentDeviceStatusEvent + var deviceIDs []uint64 + if err := service.DB.WithContext(ctx).Model(&models.AgentDevice{}). + Where("status = ? AND COALESCE(last_heartbeat_at, created_at) < ?", models.DeviceStatusOnline, cutoff). + Order("id").Pluck("id", &deviceIDs).Error; err != nil { + return 0, internalError(err) + } + if len(deviceIDs) == 0 { + return 0, nil + } err := service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - var deviceIDs []uint64 - if err := tx.Model(&models.AgentDevice{}). - Where("status = ? AND COALESCE(last_heartbeat_at, created_at) < ?", models.DeviceStatusOnline, cutoff). - Pluck("id", &deviceIDs).Error; err != nil { - return internalError(err) + // Lock primary keys in deterministic order, not the mutable status index: + // a status-range FOR UPDATE can deadlock concurrent scans on gap locks. + // Candidate selection is only a hint; revalidate the current locked row. + for _, deviceID := range deviceIDs { + var device models.AgentDevice + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, deviceID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + continue + } + return internalError(err) + } + last := device.CreatedAt + if device.LastHeartbeatAt != nil { + last = *device.LastHeartbeatAt + } + if device.Status != models.DeviceStatusOnline || !last.Before(cutoff) { + continue + } + // Existing Admin reset paths lock task -> device, while Agent paths + // lock device -> task. Never wait for a task while holding this device: + // contention aborts this atomic scan and the next scan retries it. + if err := lockOfflineTasksNowait(tx, device.ID); err != nil { + return err + } + result := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}). + Where("device_id = ? AND status = ?", device.ID, models.TaskStatusRunning). + Updates(map[string]any{ + "status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"), "device_run_slot": gorm.Expr("NULL"), + "lease_expires_at": gorm.Expr("NULL"), "error_code": "DEVICE_OFFLINE", + "error_message": "设备心跳超时，任务失败且不会自动重试", "finished_at": now, + }) + if result.Error != nil { + return internalError(result.Error) + } + failed, unknown, err := markOfflinePurchaseTasks(tx, []uint64{device.ID}, now) + if err != nil { + return err + } + event := models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusOffline, Reason: "heartbeat_timeout", LastHeartbeatAt: device.LastHeartbeatAt, FailedTaskCount: result.RowsAffected + failed, OrderResultUnknownCount: unknown} + result = tx.Model(&models.AgentDevice{}).Where("id = ? AND status = ?", device.ID, models.DeviceStatusOnline). + Update("status", models.DeviceStatusOffline) + if result.Error != nil { + return internalError(result.Error) + } + if err := insertDeviceStatusEvent(tx, &event); err != nil { + return err + } + events = append(events, event) + changed += result.RowsAffected } - if len(deviceIDs) == 0 { - return nil - } - if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}). - Where("device_id IN ? AND status = ?", deviceIDs, models.TaskStatusRunning). - Updates(map[string]any{ - "status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"), "device_run_slot": gorm.Expr("NULL"), - "lease_expires_at": gorm.Expr("NULL"), "error_code": "DEVICE_OFFLINE", - "error_message": "设备心跳超时，任务失败且不会自动重试", "finished_at": now, - }).Error; err != nil { - return internalError(err) - } - if err := markOfflinePurchaseTasks(tx, deviceIDs, now); err != nil { - return err - } - result := tx.Model(&models.AgentDevice{}).Where("id IN ? AND status = ?", deviceIDs, models.DeviceStatusOnline). - Update("status", models.DeviceStatusOffline) - if result.Error != nil { - return internalError(result.Error) - } - changed = result.RowsAffected return nil }) + if err != nil { + return 0, err + } + for _, event := range events { + logDeviceStatusEvent(event) + } return changed, err } -func markOfflinePurchaseTasks(tx *gorm.DB, deviceIDs []uint64, now time.Time) error { - if err := markOfflinePurchaseStatus( +func lockOfflineTasksNowait(tx *gorm.DB, deviceID uint64) error { + var collectionIDs, purchaseIDs []uint64 + locking := clause.Locking{Strength: "UPDATE", Options: "NOWAIT"} + if err := tx.Model(&models.CollectionTask{}).Clauses(locking). + Where("device_id = ? AND status = ?", deviceID, models.TaskStatusRunning). + Order("id").Pluck("id", &collectionIDs).Error; err != nil { + return internalError(err) + } + if err := tx.Model(&models.PurchaseTask{}).Clauses(locking). + Where("device_id = ? AND status IN ?", deviceID, []string{models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusOrderSubmitStarted}). + Order("id").Pluck("id", &purchaseIDs).Error; err != nil { + return internalError(err) + } + return nil +} + +func markOfflinePurchaseTasks(tx *gorm.DB, deviceIDs []uint64, now time.Time) (int64, int64, error) { + failed, err := markOfflinePurchaseStatus( tx, deviceIDs, models.PurchaseTaskStatusRunning, models.PurchaseTaskStatusFailed, "failed", "设备心跳超时，采购任务失败且不会自动重试", now, - ); err != nil { - return err + ) + if err != nil { + return 0, 0, err } - return markOfflinePurchaseStatus( + unknown, err := markOfflinePurchaseStatus( tx, deviceIDs, models.PurchaseTaskStatusOrderSubmitStarted, models.PurchaseTaskStatusOrderResultUnknown, "order_result_unknown", "设备心跳超时，订单结果未知，请人工核对，禁止自动重试", now, ) + return failed, unknown, err } func markOfflinePurchaseStatus( @@ -178,15 +257,15 @@ func markOfflinePurchaseStatus( resultType string, errorMessage string, now time.Time, -) error { +) (int64, error) { var taskIDs []uint64 if err := tx.Model(&models.PurchaseTask{}). Where("device_id IN ? AND status = ?", deviceIDs, fromStatus). Pluck("id", &taskIDs).Error; err != nil { - return internalError(err) + return 0, internalError(err) } if len(taskIDs) == 0 { - return nil + return 0, nil } errorCode := "DEVICE_OFFLINE" if err := tx.Model(&models.PurchaseTaskAttempt{}). @@ -195,7 +274,7 @@ func markOfflinePurchaseStatus( "status": models.PurchaseAttemptStatusFailed, "result_type": resultType, "error_code": errorCode, "error_message": errorMessage, "finished_at": now, }).Error; err != nil { - return internalError(err) + return 0, internalError(err) } updates := map[string]any{ "status": toStatus, "device_run_slot": gorm.Expr("NULL"), "account_run_slot": gorm.Expr("NULL"), @@ -205,12 +284,13 @@ func markOfflinePurchaseStatus( if toStatus == models.PurchaseTaskStatusFailed { updates["active_slot"] = gorm.Expr("NULL") } - if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}). + result := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.PurchaseTask{}). Where("id IN ? AND status = ?", taskIDs, fromStatus). - Updates(updates).Error; err != nil { - return internalError(err) + Updates(updates) + if result.Error != nil { + return 0, internalError(result.Error) } - return nil + return result.RowsAffected, nil } func RunOfflineMonitor(ctx context.Context, service *Service, scanInterval, threshold time.Duration, onError func(error)) { diff --git a/server/app/goauto/device/service.go b/server/app/goauto/device/service.go index f50adfc..2543b1e 100644 --- a/server/app/goauto/device/service.go +++ b/server/app/goauto/device/service.go @@ -16,6 +16,7 @@ import ( "github.com/google/uuid" "gorm.io/gorm" + "gorm.io/gorm/clause" ) const DefaultHeartbeatIntervalSeconds = 15 @@ -305,7 +306,15 @@ func (service *Service) RevokeToken(ctx context.Context, deviceID uint64) error func (service *Service) deactivate(ctx context.Context, deviceID uint64, revoke bool) error { now := service.Now() - return service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var event *models.AgentDeviceStatusEvent + err := service.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var device models.AgentDevice + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, deviceID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return &ServiceError{Code: CodeDeviceNotFound, Message: "设备不存在", Retryable: false} + } + return internalError(err) + } updates := map[string]any{"status": models.DeviceStatusDisabled} if revoke { updates["token_revoked_at"] = now @@ -324,17 +333,26 @@ func (service *Service) deactivate(ctx context.Context, deviceID uint64, revoke } } errorCode, errorMessage := CodeDeviceDisabled, "设备已由管理员停用" - if err := tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}). + result = tx.Session(&gorm.Session{SkipHooks: true}).Model(&models.CollectionTask{}). Where("device_id = ? AND status = ?", deviceID, models.TaskStatusRunning). Updates(map[string]any{ "status": models.TaskStatusFailed, "active_slot": gorm.Expr("NULL"), "device_run_slot": gorm.Expr("NULL"), "lease_expires_at": gorm.Expr("NULL"), "error_code": errorCode, "error_message": errorMessage, "finished_at": now, - }).Error; err != nil { - return internalError(err) + }) + if result.Error != nil { + return internalError(result.Error) + } + if device.Status != models.DeviceStatusDisabled { + event = &models.AgentDeviceStatusEvent{DeviceID: device.ID, OccurredAt: now, FromStatus: device.Status, ToStatus: models.DeviceStatusDisabled, Reason: "disabled", LastHeartbeatAt: device.LastHeartbeatAt, FailedTaskCount: result.RowsAffected} + return insertDeviceStatusEvent(tx, event) } return nil }) + if err == nil && event != nil { + logDeviceStatusEvent(*event) + } + return err } func normalizeRegisterRequest(request RegisterRequest) RegisterRequest { diff --git a/server/app/goauto/migrations/migrate.go b/server/app/goauto/migrations/migrate.go index 1a09f02..df74c68 100644 --- a/server/app/goauto/migrations/migrate.go +++ b/server/app/goauto/migrations/migrate.go @@ -30,6 +30,8 @@ LIMIT 1` func MigratedModels() []any { return []any{ &models.AgentDevice{}, + &models.AgentHeartbeatLog{}, + &models.AgentDeviceStatusEvent{}, &models.AgentAppRelease{}, &models.AgentAppReleaseSetting{}, &models.PDDProduct{}, diff --git a/server/app/goauto/models/agent_diagnostics.go b/server/app/goauto/models/agent_diagnostics.go new file mode 100644 index 0000000..e1f99e0 --- /dev/null +++ b/server/app/goauto/models/agent_diagnostics.go @@ -0,0 +1,33 @@ +package models + +import "time" + +// AgentHeartbeatLog is diagnostic history, never a device liveness authority. +type AgentHeartbeatLog struct { + ID uint64 `gorm:"primaryKey;autoIncrement"` + DeviceID uint64 `gorm:"not null;index:ix_heartbeat_device_received,priority:1"` + ReceivedAt time.Time `gorm:"not null;index:ix_heartbeat_device_received,priority:2;index:ix_heartbeat_received"` + RequestID string `gorm:"size:36;not null"` + CurrentTaskID *uint64 + RunningTaskID *uint64 + Busy bool `gorm:"not null"` + AgentVersion string `gorm:"size:32;not null"` + ClientReportJSON *string `gorm:"type:text"` + ClientReportStatus string `gorm:"size:32;not null;check:ck_heartbeat_report_status,client_report_status IN ('none','accepted','client_report_invalid')"` +} + +func (AgentHeartbeatLog) TableName() string { return "agent_heartbeat_log" } + +type AgentDeviceStatusEvent struct { + ID uint64 `gorm:"primaryKey;autoIncrement"` + DeviceID uint64 `gorm:"not null"` + OccurredAt time.Time `gorm:"not null;index:ix_device_event_occurred"` + FromStatus string `gorm:"size:16;not null"` + ToStatus string `gorm:"size:16;not null"` + Reason string `gorm:"size:32;not null"` + LastHeartbeatAt *time.Time + FailedTaskCount int64 `gorm:"not null"` + OrderResultUnknownCount int64 `gorm:"not null"` +} + +func (AgentDeviceStatusEvent) TableName() string { return "agent_device_status_event" } diff --git a/server/cmd/api/server.go b/server/cmd/api/server.go index d3ec9b8..510f132 100644 --- a/server/cmd/api/server.go +++ b/server/cmd/api/server.go @@ -129,6 +129,9 @@ func run() error { defer stopOfflineMonitors() for _, db := range sdk.Runtime.GetDb() { service := goautodevice.NewService(db) + go goautodevice.RunDiagnosticsCleanup(offlineMonitorContext, service, + func(err error) { log.Error("device diagnostics cleanup failed") }, + ) go goautopurchase.RunFailureSnapshotCleanup( offlineMonitorContext, goautopurchase.NewService(db), time.Hour, func(err error) { log.Error("purchase failure snapshot cleanup failed") }, diff --git a/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics.go b/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics.go new file mode 100644 index 0000000..ce1f569 --- /dev/null +++ b/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics.go @@ -0,0 +1,29 @@ +package version_local + +import ( + "runtime" + + "go-admin/app/goauto/models" + "go-admin/cmd/migrate/migration" + common "go-admin/common/models" + + "gorm.io/gorm" +) + +func init() { + _, file, _, _ := runtime.Caller(0) + migration.Migrate.SetVersion(migration.GetFilename(file), MigrateAgentDiagnostics) +} + +// MigrateAgentDiagnostics only adds #378 diagnostic tables. MySQL DDL commits +// implicitly: idempotent table creation allows recovery after a partial up. +func MigrateAgentDiagnostics(db *gorm.DB, version string) error { + for _, model := range []any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}} { + if !db.Migrator().HasTable(model) { + if err := db.Migrator().CreateTable(model); err != nil { + return err + } + } + } + return db.Where("version = ?", version).FirstOrCreate(&common.Migration{Version: version}).Error +} diff --git a/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics_test.go b/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics_test.go new file mode 100644 index 0000000..c4c8b36 --- /dev/null +++ b/server/cmd/migrate/migration/version-local/1791600000000_agent_diagnostics_test.go @@ -0,0 +1,48 @@ +package version_local + +import ( + "go-admin/app/goauto/models" + common "go-admin/common/models" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "testing" +) + +func TestAgentDiagnosticsMigrationAppendAndRollback(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&common.Migration{}); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err := MigrateAgentDiagnostics(db, "test378"); err != nil { + t.Fatal(err) + } + } + for _, model := range []any{&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}} { + if !db.Migrator().HasTable(model) { + t.Fatalf("missing table %T", model) + } + } + for _, idx := range []struct { + model any + name string + }{{&models.AgentHeartbeatLog{}, "ix_heartbeat_device_received"}, {&models.AgentHeartbeatLog{}, "ix_heartbeat_received"}, {&models.AgentDeviceStatusEvent{}, "ix_device_event_occurred"}} { + if !db.Migrator().HasIndex(idx.model, idx.name) { + t.Fatalf("missing index %s", idx.name) + } + } + // Existing migration runner only supports up. Rollback is an explicit, + // operator-authorized removal of these two new diagnostic tables/version. + if err := db.Migrator().DropTable(&models.AgentHeartbeatLog{}, &models.AgentDeviceStatusEvent{}); err != nil { + t.Fatal(err) + } + if err := db.Where("version = ?", "test378").Delete(&common.Migration{}).Error; err != nil { + t.Fatal(err) + } + if err := MigrateAgentDiagnostics(db, "test378"); err != nil { + t.Fatal(err) + } +}