diff --git a/controller/system_task.go b/controller/system_task.go new file mode 100644 index 00000000..cd85829c --- /dev/null +++ b/controller/system_task.go @@ -0,0 +1,96 @@ +package controller + +import ( + "net/http" + "strconv" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + "github.com/QuantumNous/new-api/service" + + "github.com/gin-gonic/gin" +) + +func CreateLogCleanupSystemTask(c *gin.Context) { + targetTimestamp, _ := strconv.ParseInt(c.Query("target_timestamp"), 10, 64) + if targetTimestamp == 0 { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "target timestamp is required", + }) + return + } + + task, err := service.StartLogCleanupTask(targetTimestamp) + if err != nil { + common.ApiError(c, err) + return + } + + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": task.ToResponse(), + }) +} + +func GetCurrentSystemTask(c *gin.Context) { + taskType := c.Query("type") + if taskType == "" { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "type is required", + }) + return + } + + task, err := model.GetActiveSystemTask(taskType) + if err != nil { + common.ApiError(c, err) + return + } + if task == nil { + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": nil, + }) + return + } + + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": task.ToResponse(), + }) +} + +func GetSystemTask(c *gin.Context) { + taskID := c.Param("task_id") + if taskID == "" { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "task id is required", + }) + return + } + + task, err := model.GetSystemTaskByTaskID(taskID) + if err != nil { + common.ApiError(c, err) + return + } + if task == nil { + c.JSON(http.StatusNotFound, gin.H{ + "success": false, + "message": "task not found", + }) + return + } + + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": task.ToResponse(), + }) +} diff --git a/main.go b/main.go index 3361b8ce..634746f7 100644 --- a/main.go +++ b/main.go @@ -119,6 +119,9 @@ func main() { // Subscription quota reset task (daily/weekly/monthly/custom) service.StartSubscriptionQuotaResetTask() + // Persistent system maintenance task runner + service.StartSystemTaskRunner() + // Wire task polling adaptor factory (breaks service -> relay import cycle) service.GetTaskAdaptorFunc = func(platform constant.TaskPlatform) service.TaskPollingAdaptor { a := relay.GetTaskAdaptor(platform) diff --git a/model/log.go b/model/log.go index 3f6db37c..720bb6c8 100644 --- a/model/log.go +++ b/model/log.go @@ -666,21 +666,25 @@ func SumUsedToken(logType int, startTimestamp int64, endTimestamp int64, modelNa return token } -func DeleteOldLog(ctx context.Context, targetTimestamp int64, limit int) (int64, error) { +func CountOldLog(ctx context.Context, targetTimestamp int64) (int64, error) { + var total int64 + if err := LOG_DB.WithContext(ctx).Model(&Log{}).Where("created_at < ?", targetTimestamp).Count(&total).Error; err != nil { + return 0, err + } + return total, nil +} + +func DeleteOldLogBatch(ctx context.Context, targetTimestamp int64, limit int) (int64, error) { + if limit <= 0 { + limit = 100 + } + if nil != ctx.Err() { + return 0, ctx.Err() + } + if common.UsingLogDatabase(common.DatabaseTypeClickHouse) { - if limit <= 0 { - limit = 100 - } - - var total int64 = 0 - - for { - if nil != ctx.Err() { - return total, ctx.Err() - } - - var batchCount int64 - if err := LOG_DB.WithContext(ctx).Raw(` + var batchCount int64 + if err := LOG_DB.WithContext(ctx).Raw(` SELECT count() FROM ( SELECT created_at, request_id FROM logs @@ -688,13 +692,13 @@ SELECT count() FROM ( ORDER BY created_at ASC, request_id ASC LIMIT ? )`, targetTimestamp, limit).Scan(&batchCount).Error; err != nil { - return total, err - } - if batchCount == 0 { - break - } + return 0, err + } + if batchCount == 0 { + return 0, nil + } - if err := LOG_DB.WithContext(ctx).Exec(` + if err := LOG_DB.WithContext(ctx).Exec(` ALTER TABLE logs DELETE WHERE (created_at, request_id) IN ( SELECT created_at, request_id FROM logs @@ -702,17 +706,22 @@ ALTER TABLE logs DELETE WHERE (created_at, request_id) IN ( ORDER BY created_at ASC, request_id ASC LIMIT ? ) SETTINGS mutations_sync = 1`, targetTimestamp, limit).Error; err != nil { - return total, err - } - - total += batchCount - - if batchCount < int64(limit) { - break - } + return 0, err } - return total, nil + return batchCount, nil + } + + result := LOG_DB.WithContext(ctx).Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{}) + if nil != result.Error { + return 0, result.Error + } + return result.RowsAffected, nil +} + +func DeleteOldLog(ctx context.Context, targetTimestamp int64, limit int) (int64, error) { + if limit <= 0 { + limit = 100 } var total int64 = 0 @@ -722,14 +731,14 @@ ALTER TABLE logs DELETE WHERE (created_at, request_id) IN ( return total, ctx.Err() } - result := LOG_DB.Where("created_at < ?", targetTimestamp).Limit(limit).Delete(&Log{}) - if nil != result.Error { - return total, result.Error + rowsAffected, err := DeleteOldLogBatch(ctx, targetTimestamp, limit) + if nil != err { + return total, err } - total += result.RowsAffected + total += rowsAffected - if result.RowsAffected < int64(limit) { + if rowsAffected < int64(limit) { break } } diff --git a/model/main.go b/model/main.go index eea7221b..f886d00b 100644 --- a/model/main.go +++ b/model/main.go @@ -294,6 +294,7 @@ func migrateDB() error { &CustomOAuthProvider{}, &UserOAuthBinding{}, &PerfMetric{}, + &SystemTask{}, ) if err != nil { return err @@ -343,6 +344,7 @@ func migrateDBFast() error { {&CustomOAuthProvider{}, "CustomOAuthProvider"}, {&UserOAuthBinding{}, "UserOAuthBinding"}, {&PerfMetric{}, "PerfMetric"}, + {&SystemTask{}, "SystemTask"}, } // 动态计算migration数量,确保errChan缓冲区足够大 errChan := make(chan error, len(migrations)) diff --git a/model/system_task.go b/model/system_task.go new file mode 100644 index 00000000..21ee3983 --- /dev/null +++ b/model/system_task.go @@ -0,0 +1,275 @@ +package model + +import ( + "errors" + + "github.com/QuantumNous/new-api/common" + + "gorm.io/gorm" +) + +type SystemTaskStatus string + +const ( + SystemTaskStatusPending SystemTaskStatus = "pending" + SystemTaskStatusRunning SystemTaskStatus = "running" + SystemTaskStatusSucceeded SystemTaskStatus = "succeeded" + SystemTaskStatusFailed SystemTaskStatus = "failed" + + SystemTaskTypeLogCleanup = "log_cleanup" +) + +var ErrSystemTaskLockLost = errors.New("system task lock lost") + +type SystemTask struct { + ID int64 `json:"id" gorm:"primary_key;AUTO_INCREMENT"` + TaskID string `json:"task_id" gorm:"type:varchar(64);uniqueIndex"` + Type string `json:"type" gorm:"type:varchar(64);index"` + Status SystemTaskStatus `json:"status" gorm:"type:varchar(32);index"` + ActiveKey *string `json:"active_key,omitempty" gorm:"type:varchar(64);uniqueIndex"` + Payload string `json:"payload" gorm:"type:text"` + State string `json:"state" gorm:"type:text"` + Result string `json:"result" gorm:"type:text"` + Error string `json:"error" gorm:"type:text"` + LockedBy string `json:"locked_by" gorm:"type:varchar(128);index"` + LockedUntil int64 `json:"locked_until" gorm:"bigint;index"` + CreatedAt int64 `json:"created_at" gorm:"bigint;index"` + UpdatedAt int64 `json:"updated_at" gorm:"bigint;index"` +} + +type SystemTaskResponse struct { + ID int64 `json:"id"` + TaskID string `json:"task_id"` + Type string `json:"type"` + Status SystemTaskStatus `json:"status"` + ActiveKey string `json:"active_key,omitempty"` + Payload any `json:"payload"` + State any `json:"state"` + Result any `json:"result"` + Error string `json:"error"` + LockedBy string `json:"locked_by"` + LockedUntil int64 `json:"locked_until"` + CreatedAt int64 `json:"created_at"` + UpdatedAt int64 `json:"updated_at"` +} + +func (task *SystemTask) BeforeCreate(_ *gorm.DB) error { + now := common.GetTimestamp() + if task.CreatedAt == 0 { + task.CreatedAt = now + } + if task.UpdatedAt == 0 { + task.UpdatedAt = now + } + return nil +} + +func GenerateSystemTaskID() (string, error) { + key, err := common.GenerateRandomCharsKey(32) + if err != nil { + return "", err + } + return "systask_" + key, nil +} + +func CreateSystemTask(taskType string, activeKey string, payload any, state any) (*SystemTask, error) { + taskID, err := GenerateSystemTaskID() + if err != nil { + return nil, err + } + payloadText, err := marshalSystemTaskJSON(payload) + if err != nil { + return nil, err + } + stateText, err := marshalSystemTaskJSON(state) + if err != nil { + return nil, err + } + + task := &SystemTask{ + TaskID: taskID, + Type: taskType, + Status: SystemTaskStatusPending, + Payload: payloadText, + State: stateText, + } + if activeKey != "" { + task.ActiveKey = &activeKey + } + + if err := DB.Create(task).Error; err != nil { + return nil, err + } + return task, nil +} + +func GetSystemTaskByTaskID(taskID string) (*SystemTask, error) { + var task SystemTask + if err := DB.Where("task_id = ?", taskID).First(&task).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + return nil, err + } + return &task, nil +} + +func GetActiveSystemTask(taskType string) (*SystemTask, error) { + var task SystemTask + err := DB.Where("type = ? AND active_key IS NOT NULL", taskType). + Where("status IN ?", activeSystemTaskStatuses()). + Order("id desc"). + First(&task).Error + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + return nil, err + } + return &task, nil +} + +func FindRunnableSystemTasks(taskType string, now int64, limit int) ([]*SystemTask, error) { + var tasks []*SystemTask + if limit <= 0 { + limit = 1 + } + err := DB.Where("type = ? AND status IN ? AND (locked_until = 0 OR locked_until < ?)", taskType, activeSystemTaskStatuses(), now). + Order("id asc"). + Limit(limit). + Find(&tasks).Error + return tasks, err +} + +func ClaimSystemTask(id int64, taskType string, runnerID string, lockUntil int64) (*SystemTask, bool, error) { + now := common.GetTimestamp() + result := DB.Model(&SystemTask{}). + Where("id = ? AND type = ? AND status IN ? AND (locked_until = 0 OR locked_until < ? OR locked_by = ?)", id, taskType, activeSystemTaskStatuses(), now, runnerID). + Updates(map[string]any{ + "status": SystemTaskStatusRunning, + "locked_by": runnerID, + "locked_until": lockUntil, + "updated_at": now, + }) + if result.Error != nil { + return nil, false, result.Error + } + if result.RowsAffected == 0 { + return nil, false, nil + } + + var task SystemTask + if err := DB.Where("id = ?", id).First(&task).Error; err != nil { + return nil, false, err + } + return &task, true, nil +} + +func UpdateSystemTaskState(taskID string, lockedBy string, state any, lockUntil int64) error { + stateText, err := marshalSystemTaskJSON(state) + if err != nil { + return err + } + result := DB.Model(&SystemTask{}). + Where("task_id = ? AND status = ? AND locked_by = ?", taskID, SystemTaskStatusRunning, lockedBy). + Updates(map[string]any{ + "state": stateText, + "locked_until": lockUntil, + "updated_at": common.GetTimestamp(), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return ErrSystemTaskLockLost + } + return nil +} + +func FinishSystemTask(taskID string, lockedBy string, status SystemTaskStatus, resultPayload any, errorMessage string) error { + resultText, err := marshalSystemTaskJSON(resultPayload) + if err != nil { + return err + } + result := DB.Model(&SystemTask{}). + Where("task_id = ? AND status = ? AND locked_by = ?", taskID, SystemTaskStatusRunning, lockedBy). + Updates(map[string]any{ + "status": status, + "active_key": nil, + "result": resultText, + "error": errorMessage, + "locked_by": "", + "locked_until": 0, + "updated_at": common.GetTimestamp(), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return ErrSystemTaskLockLost + } + return nil +} + +func (task *SystemTask) DecodePayload(v any) error { + return decodeSystemTaskJSONString(task.Payload, v) +} + +func (task *SystemTask) DecodeState(v any) error { + return decodeSystemTaskJSONString(task.State, v) +} + +func (task *SystemTask) ToResponse() SystemTaskResponse { + activeKey := "" + if task.ActiveKey != nil { + activeKey = *task.ActiveKey + } + return SystemTaskResponse{ + ID: task.ID, + TaskID: task.TaskID, + Type: task.Type, + Status: task.Status, + ActiveKey: activeKey, + Payload: decodeSystemTaskJSONValue(task.Payload), + State: decodeSystemTaskJSONValue(task.State), + Result: decodeSystemTaskJSONValue(task.Result), + Error: task.Error, + LockedBy: task.LockedBy, + LockedUntil: task.LockedUntil, + CreatedAt: task.CreatedAt, + UpdatedAt: task.UpdatedAt, + } +} + +func activeSystemTaskStatuses() []string { + return []string{string(SystemTaskStatusPending), string(SystemTaskStatusRunning)} +} + +func marshalSystemTaskJSON(v any) (string, error) { + if v == nil { + return "", nil + } + data, err := common.Marshal(v) + if err != nil { + return "", err + } + return string(data), nil +} + +func decodeSystemTaskJSONString(data string, v any) error { + if data == "" { + return nil + } + return common.UnmarshalJsonStr(data, v) +} + +func decodeSystemTaskJSONValue(data string) any { + if data == "" { + return nil + } + var value any + if err := common.UnmarshalJsonStr(data, &value); err != nil { + return data + } + return value +} diff --git a/model/system_task_test.go b/model/system_task_test.go new file mode 100644 index 00000000..68496e24 --- /dev/null +++ b/model/system_task_test.go @@ -0,0 +1,83 @@ +package model + +import ( + "testing" + + "github.com/QuantumNous/new-api/common" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type testSystemTaskPayload struct { + TargetTimestamp int64 `json:"target_timestamp"` + BatchSize int `json:"batch_size"` +} + +type testSystemTaskState struct { + Total int64 `json:"total"` + Processed int64 `json:"processed"` + Progress int `json:"progress"` + Remaining int64 `json:"remaining"` +} + +func TestSystemTaskActiveKeyIsReleasedOnFinish(t *testing.T) { + truncateTables(t) + + payload := testSystemTaskPayload{TargetTimestamp: 1000, BatchSize: 100} + state := testSystemTaskState{} + task, err := CreateSystemTask(SystemTaskTypeLogCleanup, SystemTaskTypeLogCleanup, payload, state) + require.NoError(t, err) + + var decodedPayload testSystemTaskPayload + require.NoError(t, task.DecodePayload(&decodedPayload)) + assert.Equal(t, payload, decodedPayload) + + _, err = CreateSystemTask(SystemTaskTypeLogCleanup, SystemTaskTypeLogCleanup, payload, state) + require.Error(t, err) + + activeTask, err := GetActiveSystemTask(SystemTaskTypeLogCleanup) + require.NoError(t, err) + require.NotNil(t, activeTask) + assert.Equal(t, task.TaskID, activeTask.TaskID) + + runnerID := "runner-a" + claimedTask, claimed, err := ClaimSystemTask(task.ID, SystemTaskTypeLogCleanup, runnerID, common.GetTimestamp()+60) + require.NoError(t, err) + require.True(t, claimed) + + err = FinishSystemTask(claimedTask.TaskID, runnerID, SystemTaskStatusSucceeded, map[string]int64{"deleted_count": 0}, "") + require.NoError(t, err) + + activeTask, err = GetActiveSystemTask(SystemTaskTypeLogCleanup) + require.NoError(t, err) + require.Nil(t, activeTask) + + _, err = CreateSystemTask(SystemTaskTypeLogCleanup, SystemTaskTypeLogCleanup, payload, state) + require.NoError(t, err) +} + +func TestSystemTaskClaimRequiresExpiredLock(t *testing.T) { + truncateTables(t) + + payload := testSystemTaskPayload{TargetTimestamp: 1000, BatchSize: 100} + task, err := CreateSystemTask(SystemTaskTypeLogCleanup, SystemTaskTypeLogCleanup, payload, testSystemTaskState{}) + require.NoError(t, err) + + claimedTask, claimed, err := ClaimSystemTask(task.ID, SystemTaskTypeLogCleanup, "runner-a", common.GetTimestamp()+60) + require.NoError(t, err) + require.True(t, claimed) + + _, claimed, err = ClaimSystemTask(task.ID, SystemTaskTypeLogCleanup, "runner-b", common.GetTimestamp()+60) + require.NoError(t, err) + require.False(t, claimed) + + require.NoError(t, DB.Model(claimedTask).Updates(map[string]any{ + "locked_until": common.GetTimestamp() - 1, + }).Error) + + claimedTask, claimed, err = ClaimSystemTask(task.ID, SystemTaskTypeLogCleanup, "runner-b", common.GetTimestamp()+60) + require.NoError(t, err) + require.True(t, claimed) + assert.Equal(t, "runner-b", claimedTask.LockedBy) +} diff --git a/model/task_cas_test.go b/model/task_cas_test.go index 25928220..f8288656 100644 --- a/model/task_cas_test.go +++ b/model/task_cas_test.go @@ -48,6 +48,7 @@ func TestMain(m *testing.M) { &UserSubscription{}, &UserOAuthBinding{}, &PerfMetric{}, + &SystemTask{}, ); err != nil { panic("failed to migrate: " + err.Error()) } @@ -71,6 +72,7 @@ func truncateTables(t *testing.T) { DB.Exec("DELETE FROM user_subscriptions") DB.Exec("DELETE FROM user_oauth_bindings") DB.Exec("DELETE FROM perf_metrics") + DB.Exec("DELETE FROM system_tasks") }) } diff --git a/router/api-router.go b/router/api-router.go index 7fac06f4..e16fd285 100644 --- a/router/api-router.go +++ b/router/api-router.go @@ -304,7 +304,7 @@ func SetApiRouter(router *gin.Engine) { } logRoute := apiRouter.Group("/log") logRoute.GET("/", middleware.AdminAuth(), controller.GetAllLogs) - logRoute.DELETE("/", middleware.AdminAuth(), controller.DeleteHistoryLogs) + logRoute.DELETE("/", middleware.RootAuth(), controller.DeleteHistoryLogs) logRoute.GET("/stat", middleware.AdminAuth(), controller.GetLogsStat) logRoute.GET("/self/stat", middleware.UserAuth(), controller.GetLogsSelfStat) logRoute.GET("/channel_affinity_usage_cache", middleware.AdminAuth(), controller.GetChannelAffinityUsageCacheStats) @@ -312,6 +312,14 @@ func SetApiRouter(router *gin.Engine) { logRoute.GET("/self", middleware.UserAuth(), controller.GetUserLogs) logRoute.GET("/self/search", middleware.UserAuth(), middleware.SearchRateLimit(), controller.SearchUserLogs) + systemTaskRoute := apiRouter.Group("/system-task") + systemTaskRoute.Use(middleware.RootAuth()) + { + systemTaskRoute.POST("/log-cleanup", controller.CreateLogCleanupSystemTask) + systemTaskRoute.GET("/current", controller.GetCurrentSystemTask) + systemTaskRoute.GET("/:task_id", controller.GetSystemTask) + } + dataRoute := apiRouter.Group("/data") dataRoute.GET("/", middleware.AdminAuth(), controller.GetAllQuotaDates) dataRoute.GET("/users", middleware.AdminAuth(), controller.GetQuotaDatesByUser) diff --git a/service/system_task.go b/service/system_task.go new file mode 100644 index 00000000..0661acd6 --- /dev/null +++ b/service/system_task.go @@ -0,0 +1,249 @@ +package service + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/logger" + "github.com/QuantumNous/new-api/model" + + "github.com/bytedance/gopkg/util/gopool" +) + +const ( + systemTaskRunnerTickInterval = time.Second + systemTaskLockTTL = 60 * time.Second + logCleanupBatchSize = 100 +) + +type LogCleanupPayload struct { + TargetTimestamp int64 `json:"target_timestamp"` + BatchSize int `json:"batch_size"` +} + +type LogCleanupState struct { + Total int64 `json:"total"` + Processed int64 `json:"processed"` + Progress int `json:"progress"` + Remaining int64 `json:"remaining"` +} + +type LogCleanupResult struct { + DeletedCount int64 `json:"deleted_count"` +} + +var systemTaskRunnerOnce sync.Once + +func StartSystemTaskRunner() { + systemTaskRunnerOnce.Do(func() { + if !common.IsMasterNode { + return + } + + runnerID := fmt.Sprintf("%s-%s", common.NodeName, common.GetRandomString(8)) + gopool.Go(func() { + logger.LogInfo(context.Background(), fmt.Sprintf("system task runner started: runner=%s tick=%s", runnerID, systemTaskRunnerTickInterval)) + + ticker := time.NewTicker(systemTaskRunnerTickInterval) + defer ticker.Stop() + + runSystemTaskRunnerOnce(runnerID) + for range ticker.C { + runSystemTaskRunnerOnce(runnerID) + } + }) + }) +} + +func StartLogCleanupTask(targetTimestamp int64) (*model.SystemTask, error) { + if targetTimestamp <= 0 { + return nil, errors.New("target timestamp is required") + } + + activeTask, err := model.GetActiveSystemTask(model.SystemTaskTypeLogCleanup) + if err != nil { + return nil, err + } + if activeTask != nil { + return activeTask, nil + } + + payload := LogCleanupPayload{ + TargetTimestamp: targetTimestamp, + BatchSize: logCleanupBatchSize, + } + state := LogCleanupState{} + task, err := model.CreateSystemTask(model.SystemTaskTypeLogCleanup, model.SystemTaskTypeLogCleanup, payload, state) + if err != nil { + activeTask, activeErr := model.GetActiveSystemTask(model.SystemTaskTypeLogCleanup) + if activeErr == nil && activeTask != nil { + return activeTask, nil + } + return nil, err + } + return task, nil +} + +func runSystemTaskRunnerOnce(runnerID string) { + now := common.GetTimestamp() + tasks, err := model.FindRunnableSystemTasks(model.SystemTaskTypeLogCleanup, now, 1) + if err != nil { + logger.LogWarn(context.Background(), fmt.Sprintf("system task runner query failed: %v", err)) + return + } + + for _, task := range tasks { + claimedTask, claimed, err := model.ClaimSystemTask(task.ID, model.SystemTaskTypeLogCleanup, runnerID, systemTaskLockUntil()) + if err != nil { + logger.LogWarn(context.Background(), fmt.Sprintf("system task claim failed: %v", err)) + continue + } + if !claimed { + continue + } + runLogCleanupTask(context.Background(), claimedTask, runnerID) + } +} + +func runLogCleanupTask(ctx context.Context, task *model.SystemTask, runnerID string) { + payload := LogCleanupPayload{} + if err := task.DecodePayload(&payload); err != nil { + failSystemTask(task, runnerID, err) + return + } + if payload.TargetTimestamp <= 0 { + failSystemTask(task, runnerID, errors.New("target timestamp is required")) + return + } + if payload.BatchSize <= 0 { + payload.BatchSize = logCleanupBatchSize + } + + state := LogCleanupState{} + if err := task.DecodeState(&state); err != nil { + failSystemTask(task, runnerID, err) + return + } + + for { + remaining, err := model.CountOldLog(ctx, payload.TargetTimestamp) + if err != nil { + failSystemTask(task, runnerID, err) + return + } + syncLogCleanupStateFromRemaining(&state, remaining) + if err := model.UpdateSystemTaskState(task.TaskID, runnerID, state, systemTaskLockUntil()); err != nil { + logSystemTaskLockError(ctx, task, err) + return + } + if state.Remaining == 0 { + break + } + + // Track whether this pass deleted anything so a fresh recount that still + // reports remaining rows resumes immediately instead of waiting for the + // lock to expire. If a whole pass deletes nothing while rows remain, the + // rows cannot be removed and we fail instead of busy-looping. + progressed := false + for state.Remaining > 0 { + rowsAffected, err := model.DeleteOldLogBatch(ctx, payload.TargetTimestamp, payload.BatchSize) + if err != nil { + failSystemTask(task, runnerID, err) + return + } + if rowsAffected == 0 { + break + } + progressed = true + + state.Processed += rowsAffected + if state.Total < state.Processed { + state.Total = state.Processed + } + if state.Remaining > rowsAffected { + state.Remaining -= rowsAffected + } else { + state.Remaining = 0 + } + state.Progress = logCleanupProgress(state.Processed, state.Total) + + if err := model.UpdateSystemTaskState(task.TaskID, runnerID, state, systemTaskLockUntil()); err != nil { + logSystemTaskLockError(ctx, task, err) + return + } + } + + if !progressed { + failSystemTask(task, runnerID, errors.New("no log rows were deleted")) + return + } + } + + state.Remaining = 0 + state.Progress = 100 + if state.Total < state.Processed { + state.Total = state.Processed + } + if err := model.UpdateSystemTaskState(task.TaskID, runnerID, state, systemTaskLockUntil()); err != nil { + logSystemTaskLockError(ctx, task, err) + return + } + + result := LogCleanupResult{DeletedCount: state.Processed} + if err := model.FinishSystemTask(task.TaskID, runnerID, model.SystemTaskStatusSucceeded, result, ""); err != nil { + logSystemTaskLockError(ctx, task, err) + } +} + +func syncLogCleanupStateFromRemaining(state *LogCleanupState, remaining int64) { + if state.Total <= 0 { + state.Total = remaining + state.Processed = 0 + } else { + processedFromRemaining := state.Total - remaining + if processedFromRemaining > state.Processed { + state.Processed = processedFromRemaining + } + } + if state.Processed < 0 { + state.Processed = 0 + } + state.Remaining = remaining + state.Progress = logCleanupProgress(state.Processed, state.Total) +} + +func logCleanupProgress(processed int64, total int64) int { + if total <= 0 { + return 100 + } + if processed <= 0 { + return 0 + } + if processed >= total { + return 100 + } + return int(processed * 100 / total) +} + +func systemTaskLockUntil() int64 { + return common.GetTimestamp() + int64(systemTaskLockTTL.Seconds()) +} + +func failSystemTask(task *model.SystemTask, runnerID string, err error) { + logger.LogWarn(context.Background(), fmt.Sprintf("system task %s failed: %v", task.TaskID, err)) + if finishErr := model.FinishSystemTask(task.TaskID, runnerID, model.SystemTaskStatusFailed, nil, err.Error()); finishErr != nil { + logger.LogWarn(context.Background(), fmt.Sprintf("system task %s failed to save failure state: %v", task.TaskID, finishErr)) + } +} + +func logSystemTaskLockError(ctx context.Context, task *model.SystemTask, err error) { + if errors.Is(err, model.ErrSystemTaskLockLost) { + logger.LogWarn(ctx, fmt.Sprintf("system task %s lock lost", task.TaskID)) + return + } + logger.LogWarn(ctx, fmt.Sprintf("system task %s update failed: %v", task.TaskID, err)) +} diff --git a/service/task_billing_test.go b/service/task_billing_test.go index b8513e00..4f05300c 100644 --- a/service/task_billing_test.go +++ b/service/task_billing_test.go @@ -44,6 +44,7 @@ func TestMain(m *testing.M) { &model.Channel{}, &model.TopUp{}, &model.UserSubscription{}, + &model.SystemTask{}, ); err != nil { panic("failed to migrate: " + err.Error()) } @@ -65,6 +66,7 @@ func truncate(t *testing.T) { model.DB.Exec("DELETE FROM channels") model.DB.Exec("DELETE FROM top_ups") model.DB.Exec("DELETE FROM user_subscriptions") + model.DB.Exec("DELETE FROM system_tasks") }) } diff --git a/web/default/src/features/system-settings/api.ts b/web/default/src/features/system-settings/api.ts index 86f41792..d942634f 100644 --- a/web/default/src/features/system-settings/api.ts +++ b/web/default/src/features/system-settings/api.ts @@ -21,7 +21,9 @@ import type { ConfirmPaymentComplianceResponse, DeleteLogsResponse, FetchUpstreamRatiosRequest, + LogCleanupTask, SystemOptionsResponse, + SystemTaskResponse, UpdateOptionRequest, UpdateOptionResponse, UpstreamChannelsResponse, @@ -53,6 +55,34 @@ export async function deleteLogsBefore(targetTimestamp: number) { return res.data } +export async function startLogCleanupTask(targetTimestamp: number) { + const res = await api.post>( + '/api/system-task/log-cleanup', + null, + { + params: { target_timestamp: targetTimestamp }, + } + ) + return res.data +} + +export async function getCurrentLogCleanupTask() { + const res = await api.get>( + '/api/system-task/current', + { + params: { type: 'log_cleanup' }, + } + ) + return res.data +} + +export async function getSystemTask(taskId: string) { + const res = await api.get>( + `/api/system-task/${taskId}` + ) + return res.data +} + export async function resetModelRatios() { const res = await api.post( '/api/option/rest_model_ratio' diff --git a/web/default/src/features/system-settings/maintenance/log-settings-section.tsx b/web/default/src/features/system-settings/maintenance/log-settings-section.tsx index b23a128c..37906f13 100644 --- a/web/default/src/features/system-settings/maintenance/log-settings-section.tsx +++ b/web/default/src/features/system-settings/maintenance/log-settings-section.tsx @@ -48,6 +48,7 @@ import { } from '@/components/ui/form' import { Input } from '@/components/ui/input' import { Label } from '@/components/ui/label' +import { Progress } from '@/components/ui/progress' import { Select, SelectContent, @@ -59,7 +60,11 @@ import { import { Separator } from '@/components/ui/separator' import { Switch } from '@/components/ui/switch' import { DateTimePicker } from '@/components/datetime-picker' -import { deleteLogsBefore } from '../api' +import { + getCurrentLogCleanupTask, + getSystemTask, + startLogCleanupTask, +} from '../api' import { SettingsControlGroup, SettingsForm, @@ -69,6 +74,7 @@ import { import { SettingsPageFormActions } from '../components/settings-page-context' import { SettingsSection } from '../components/settings-section' import { useUpdateOption } from '../hooks/use-update-option' +import type { LogCleanupTask } from '../types' const logSettingsSchema = z.object({ LogConsumeEnabled: z.boolean(), @@ -127,6 +133,10 @@ const quickSelectOptions = [ }, ] +function isActiveLogCleanupTask(task: LogCleanupTask | null) { + return task?.status === 'pending' || task?.status === 'running' +} + export function LogSettingsSection({ defaultEnabled, }: LogSettingsSectionProps) { @@ -142,7 +152,10 @@ export function LogSettingsSection({ const [purgeDate, setPurgeDate] = useState(() => getDateDaysAgo(30) ) - const [isCleaning, setIsCleaning] = useState(false) + const [isStartingLogCleanup, setIsStartingLogCleanup] = useState(false) + const [logCleanupTask, setLogCleanupTask] = useState( + null + ) const [showConfirmDialog, setShowConfirmDialog] = useState(false) const [serverLogInfo, setServerLogInfo] = useState( null @@ -168,6 +181,27 @@ export function LogSettingsSection({ fetchServerLogInfo() }, [fetchServerLogInfo]) + useEffect(() => { + let cancelled = false + + async function fetchCurrentLogCleanupTask() { + try { + const res = await getCurrentLogCleanupTask() + if (!cancelled && res.success && res.data) { + setLogCleanupTask(res.data) + } + } catch { + /* ignore */ + } + } + + fetchCurrentLogCleanupTask() + + return () => { + cancelled = true + } + }, []) + const purgeTimestamp = useMemo(() => { if (!purgeDate) return null return Math.floor(purgeDate.getTime() / 1000) @@ -178,6 +212,49 @@ export function LogSettingsSection({ return formatTimestampToDate(purgeDate.getTime(), 'milliseconds') }, [purgeDate]) + const logCleanupActive = isActiveLogCleanupTask(logCleanupTask) + const logCleanupState = logCleanupTask?.state + const logCleanupProgress = Math.min( + 100, + Math.max(0, logCleanupState?.progress ?? 0) + ) + const logCleanupProcessed = logCleanupState?.processed ?? 0 + const logCleanupTotal = logCleanupState?.total ?? 0 + + useEffect(() => { + if (!logCleanupTask || !isActiveLogCleanupTask(logCleanupTask)) return + + let cancelled = false + const interval = window.setInterval(async () => { + try { + const res = await getSystemTask(logCleanupTask.task_id) + if (cancelled || !res.success || !res.data) return + + setLogCleanupTask(res.data) + if (!isActiveLogCleanupTask(res.data)) { + if (res.data.status === 'succeeded') { + const count = + res.data.result?.deleted_count ?? res.data.state?.processed ?? 0 + toast.success( + count > 0 + ? t('{{count}} log entries removed.', { count }) + : t('No log entries matched the selected time.') + ) + } else if (res.data.status === 'failed') { + toast.error(res.data.error || t('Failed to clean logs')) + } + } + } catch { + /* keep polling */ + } + }, 1000) + + return () => { + cancelled = true + window.clearInterval(interval) + } + }, [logCleanupTask?.task_id, logCleanupTask?.status, t]) + const onSubmit = async (values: LogSettingsFormValues) => { if (values.LogConsumeEnabled === defaultEnabled) return await updateOption.mutateAsync({ @@ -201,24 +278,24 @@ export function LogSettingsSection({ return } - setIsCleaning(true) + setIsStartingLogCleanup(true) try { - const res = await deleteLogsBefore(purgeTimestamp) + const res = await startLogCleanupTask(purgeTimestamp) if (!res.success) { throw new Error(res.message || t('Failed to clean logs')) } - const count = res.data ?? 0 - toast.success( - count > 0 - ? t('{{count}} log entries removed.', { count }) - : t('No log entries matched the selected time.') - ) + if (!res.data) { + throw new Error(t('Failed to clean logs')) + } + setLogCleanupTask(res.data) + setShowConfirmDialog(false) + toast.success(t('Log cleanup task started.')) } catch (error) { const message = error instanceof Error ? error.message : t('Failed to clean logs') toast.error(message) } finally { - setIsCleaning(false) + setIsStartingLogCleanup(false) } } @@ -314,11 +391,37 @@ export function LogSettingsSection({ type='button' variant='destructive' onClick={handleRequestCleanLogs} - disabled={isCleaning} + disabled={isStartingLogCleanup || logCleanupActive} > - {isCleaning ? t('Cleaning...') : t('Clean logs')} + {isStartingLogCleanup || logCleanupActive + ? t('Cleaning...') + : t('Clean logs')} + {logCleanupTask && ( +
+
+ + {t('Log cleanup progress')} + + + {logCleanupProgress}% + +
+ +
+ {t('{{processed}} of {{total}} log entries processed.', { + processed: logCleanupProcessed, + total: logCleanupTotal, + })} +
+ {logCleanupTask.status === 'failed' && logCleanupTask.error && ( +
+ {logCleanupTask.error} +
+ )} +
+ )} @@ -491,11 +594,14 @@ export function LogSettingsSection({ - + {t('Cancel')} - - {isCleaning ? t('Cleaning...') : t('Delete logs')} + + {isStartingLogCleanup ? t('Cleaning...') : t('Delete logs')} diff --git a/web/default/src/features/system-settings/types.ts b/web/default/src/features/system-settings/types.ts index c309821a..28acd48e 100644 --- a/web/default/src/features/system-settings/types.ts +++ b/web/default/src/features/system-settings/types.ts @@ -56,6 +56,56 @@ export type DeleteLogsResponse = { data?: number } +export type SystemTaskStatus = 'pending' | 'running' | 'succeeded' | 'failed' + +export type SystemTask< + TPayload = Record, + TState = Record, + TResult = Record, +> = { + id: number + task_id: string + type: string + status: SystemTaskStatus + active_key?: string + payload?: TPayload + state?: TState + result?: TResult + error?: string + locked_by?: string + locked_until?: number + created_at: number + updated_at: number +} + +export type LogCleanupTaskPayload = { + target_timestamp: number + batch_size: number +} + +export type LogCleanupTaskState = { + total: number + processed: number + progress: number + remaining: number +} + +export type LogCleanupTaskResult = { + deleted_count: number +} + +export type LogCleanupTask = SystemTask< + LogCleanupTaskPayload, + LogCleanupTaskState, + LogCleanupTaskResult +> + +export type SystemTaskResponse = { + success: boolean + message: string + data?: TTask +} + export type SiteSettings = { 'theme.frontend': string Notice: string diff --git a/web/default/src/i18n/locales/en.json b/web/default/src/i18n/locales/en.json index 519336f6..c1c0e9ca 100644 --- a/web/default/src/i18n/locales/en.json +++ b/web/default/src/i18n/locales/en.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "{{modality}} not supported", "{{modality}} supported": "{{modality}} supported", "{{n}} model(s) selected": "{{n}} model(s) selected", + "{{processed}} of {{total}} log entries processed.": "{{processed}} of {{total}} log entries processed.", "{{success}} succeeded, {{failed}} failed": "{{success}} succeeded, {{failed}} failed", "{{target}} test failed": "{{target}} test failed", "{{target}} test succeeded": "{{target}} test succeeded", @@ -2352,6 +2353,8 @@ "Locations": "Locations", "Locked": "Locked", "log": "log", + "Log cleanup progress": "Log cleanup progress", + "Log cleanup task started.": "Log cleanup task started.", "Log Details": "Log Details", "Log Directory": "Log Directory", "Log File Count": "Log File Count", diff --git a/web/default/src/i18n/locales/fr.json b/web/default/src/i18n/locales/fr.json index c14bfd26..5c096ca7 100644 --- a/web/default/src/i18n/locales/fr.json +++ b/web/default/src/i18n/locales/fr.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "{{modality}} non pris en charge", "{{modality}} supported": "{{modality}} pris en charge", "{{n}} model(s) selected": "{{n}} modèle(s) sélectionné(s)", + "{{processed}} of {{total}} log entries processed.": "{{processed}} sur {{total}} entrées de journal traitées.", "{{success}} succeeded, {{failed}} failed": "{{success}} réussi(s), {{failed}} échoué(s)", "{{target}} test failed": "Échec du test de {{target}}", "{{target}} test succeeded": "Test de {{target}} réussi", @@ -2352,6 +2353,8 @@ "Locations": "Emplacements", "Locked": "Verrouillé", "log": "journal", + "Log cleanup progress": "Progression du nettoyage des journaux", + "Log cleanup task started.": "La tâche de nettoyage des journaux a démarré.", "Log Details": "Détails du journal", "Log Directory": "Répertoire des journaux", "Log File Count": "Nombre de fichiers journaux", diff --git a/web/default/src/i18n/locales/ja.json b/web/default/src/i18n/locales/ja.json index 1d2e6cd5..7e694fe0 100644 --- a/web/default/src/i18n/locales/ja.json +++ b/web/default/src/i18n/locales/ja.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "{{modality}} はサポートされていません", "{{modality}} supported": "{{modality}} をサポート", "{{n}} model(s) selected": "{{n}} 件のモデルを選択済み", + "{{processed}} of {{total}} log entries processed.": "{{total}} 件中 {{processed}} 件のログを処理しました。", "{{success}} succeeded, {{failed}} failed": "{{success}} 件成功、{{failed}} 件失敗", "{{target}} test failed": "{{target}} のテストに失敗しました", "{{target}} test succeeded": "{{target}} のテストに成功しました", @@ -2352,6 +2353,8 @@ "Locations": "場所", "Locked": "ロック済み", "log": "ログ", + "Log cleanup progress": "ログクリーンアップの進行状況", + "Log cleanup task started.": "ログクリーンアップタスクを開始しました。", "Log Details": "ログの詳細", "Log Directory": "ログディレクトリ", "Log File Count": "ログファイル数", diff --git a/web/default/src/i18n/locales/ru.json b/web/default/src/i18n/locales/ru.json index f5e90213..ee6839b9 100644 --- a/web/default/src/i18n/locales/ru.json +++ b/web/default/src/i18n/locales/ru.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "{{modality}} не поддерживается", "{{modality}} supported": "{{modality}} поддерживается", "{{n}} model(s) selected": "Выбрано моделей: {{n}}", + "{{processed}} of {{total}} log entries processed.": "Обработано {{processed}} из {{total}} записей журнала.", "{{success}} succeeded, {{failed}} failed": "{{success}} успешно, {{failed}} с ошибкой", "{{target}} test failed": "Тест {{target}} не выполнен", "{{target}} test succeeded": "Тест {{target}} успешно выполнен", @@ -2352,6 +2353,8 @@ "Locations": "Местоположения", "Locked": "Заблокировано", "log": "записи", + "Log cleanup progress": "Ход очистки журнала", + "Log cleanup task started.": "Задача очистки журнала запущена.", "Log Details": "Детали журнала", "Log Directory": "Каталог журналов", "Log File Count": "Кол-во файлов журналов", diff --git a/web/default/src/i18n/locales/vi.json b/web/default/src/i18n/locales/vi.json index 13dadd2a..c1791f29 100644 --- a/web/default/src/i18n/locales/vi.json +++ b/web/default/src/i18n/locales/vi.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "Không hỗ trợ {{modality}}", "{{modality}} supported": "Hỗ trợ {{modality}}", "{{n}} model(s) selected": "Đã chọn {{n}} model", + "{{processed}} of {{total}} log entries processed.": "Đã xử lý {{processed}}/{{total}} mục nhật ký.", "{{success}} succeeded, {{failed}} failed": "{{success}} thành công, {{failed}} thất bại", "{{target}} test failed": "Kiểm tra {{target}} thất bại", "{{target}} test succeeded": "Kiểm tra {{target}} thành công", @@ -2352,6 +2353,8 @@ "Locations": "Vị trí", "Locked": "Đã khóa", "log": "nhật ký", + "Log cleanup progress": "Tiến trình dọn dẹp nhật ký", + "Log cleanup task started.": "Đã bắt đầu tác vụ dọn dẹp nhật ký.", "Log Details": "Chi tiết Nhật ký", "Log Directory": "Thư mục nhật ký", "Log File Count": "Số tệp nhật ký", diff --git a/web/default/src/i18n/locales/zh.json b/web/default/src/i18n/locales/zh.json index 1e5d7f44..30b92653 100644 --- a/web/default/src/i18n/locales/zh.json +++ b/web/default/src/i18n/locales/zh.json @@ -52,6 +52,7 @@ "{{modality}} not supported": "不支持 {{modality}}", "{{modality}} supported": "支持 {{modality}}", "{{n}} model(s) selected": "已选 {{n}} 个模型", + "{{processed}} of {{total}} log entries processed.": "已处理 {{processed}} / {{total}} 条日志。", "{{success}} succeeded, {{failed}} failed": "{{success}} 个成功,{{failed}} 个失败", "{{target}} test failed": "{{target}} 测试失败", "{{target}} test succeeded": "{{target}} 测试成功", @@ -2352,6 +2353,8 @@ "Locations": "位置", "Locked": "锁定", "log": "日志的完整详情", + "Log cleanup progress": "日志清理进度", + "Log cleanup task started.": "日志清理任务已启动。", "Log Details": "日志详情", "Log Directory": "日志目录", "Log File Count": "日志文件数",