diff --git a/internal/config/config.go b/internal/config/config.go index 2b69dcb..2ba9b67 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -150,6 +150,13 @@ type TaskCenterArchiveConfig struct { BatchSize int FlushSeconds int Workers int + GC TaskCenterArchiveGCConfig +} + +// TaskCenterArchiveGCConfig 只保存清理总开关。cutoff 和批量必须由一次管理预检冻结， +// 不能由多节点各自读取易漂移的 NOW()/环境变量后直接删数据。 +type TaskCenterArchiveGCConfig struct { + Enabled bool } // WeekStarConfig 保存周榜模块配置。 @@ -456,6 +463,10 @@ func Load() Config { BatchSize: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_BATCH_SIZE", "TASK_CENTER_ARCHIVE_BATCH_SIZE"}, 500), FlushSeconds: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_FLUSH_SECONDS", "TASK_CENTER_ARCHIVE_FLUSH_SECONDS"}, 60), Workers: getEnvIntAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_WORKERS", "TASK_CENTER_ARCHIVE_WORKERS"}, 2), + GC: TaskCenterArchiveGCConfig{ + // 清理涉及不可逆数据库删除；新环境必须显式开启，避免仅部署代码就开始工作。 + Enabled: getEnvBoolAny([]string{"CHATAPP_TASK_CENTER_ARCHIVE_GC_ENABLED", "TASK_CENTER_ARCHIVE_GC_ENABLED"}, false), + }, }, }, WeekStar: WeekStarConfig{ diff --git a/internal/model/taskcenter_models.go b/internal/model/taskcenter_models.go index 27bdb1c..cfcfad3 100644 --- a/internal/model/taskcenter_models.go +++ b/internal/model/taskcenter_models.go @@ -136,6 +136,54 @@ func (TaskCenterEventArchiveOutbox) TableName() string { return "task_center_event_archive_outbox" } +// TaskCenterArchiveManifest 保存 COS 对象的可验证清单。GC 只有在源对象和恢复包都达到 +// VERIFIED 后才允许删行，避免把“PUT 返回成功”错误地当成可恢复承诺。 +type TaskCenterArchiveManifest struct { + ID int64 `gorm:"column:id;primaryKey;autoIncrement"` + ObjectType string `gorm:"column:object_type;size:32;index:idx_task_center_archive_manifest_run,priority:2"` + RunID string `gorm:"column:run_id;size:64;index:idx_task_center_archive_manifest_run,priority:1"` + COSKey string `gorm:"column:cos_key;size:512;uniqueIndex:uk_task_center_archive_manifest_cos_key"` + ParentCOSKey string `gorm:"column:parent_cos_key;size:512"` + SHA256 string `gorm:"column:sha256;size:64"` + ETag string `gorm:"column:etag;size:128"` + CompressedSize int64 `gorm:"column:compressed_size"` + RowCount int `gorm:"column:row_count"` + MinOutboxID *int64 `gorm:"column:min_outbox_id"` + MaxOutboxID *int64 `gorm:"column:max_outbox_id"` + MinOccurredAt *time.Time `gorm:"column:min_occurred_at"` + MaxOccurredAt *time.Time `gorm:"column:max_occurred_at"` + VerificationStatus string `gorm:"column:verification_status;size:32"` + VerifiedAt *time.Time `gorm:"column:verified_at"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (TaskCenterArchiveManifest) TableName() string { return "task_center_archive_manifest" } + +// TaskCenterArchiveGCRun 是固定 id=1 的全局控制行。管理命令、cron tick 和 Game King +// 启用门禁都先锁这一行，从数据库层串行化多节点状态变化。 +type TaskCenterArchiveGCRun struct { + ID uint8 `gorm:"column:id;primaryKey"` + RunID string `gorm:"column:run_id;size:64"` + Status string `gorm:"column:status;size:32"` + CutoffTime *time.Time `gorm:"column:cutoff_time"` + BatchSize int `gorm:"column:batch_size"` + ScannedRows int64 `gorm:"column:scanned_rows"` + EligibleRows int64 `gorm:"column:eligible_rows"` + ProtectedRows int64 `gorm:"column:protected_rows"` + DeletedRows int64 `gorm:"column:deleted_rows"` + RecoveryObjectCount int64 `gorm:"column:recovery_object_count"` + CursorCreateTime *time.Time `gorm:"column:cursor_create_time"` + CursorID int64 `gorm:"column:cursor_id"` + LeaseOwner string `gorm:"column:lease_owner;size:160"` + LeaseUntil *time.Time `gorm:"column:lease_until"` + LastError string `gorm:"column:last_error;size:1000"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (TaskCenterArchiveGCRun) TableName() string { return "task_center_archive_gc_run" } + // TaskCenterClaimRecord 保存任务奖励领取记录。 type TaskCenterClaimRecord struct { ID int64 `gorm:"column:id;primaryKey"` diff --git a/internal/model/yumi_game_king_models.go b/internal/model/yumi_game_king_models.go index 93665d3..bd4c999 100644 --- a/internal/model/yumi_game_king_models.go +++ b/internal/model/yumi_game_king_models.go @@ -82,6 +82,21 @@ type YumiGameKingConsumeLedger struct { func (YumiGameKingConsumeLedger) TableName() string { return "yumi_game_king_consume_ledger" } +// YumiGameKingBackfillState 持久化复合游标。调用方重启后从该水位继续，GC 不依赖 +// webconsole/chatapp-cron 的内存 lastId 来判断历史事件是否已经确认。 +type YumiGameKingBackfillState struct { + ActivityID int64 `gorm:"column:activity_id;primaryKey"` + LastOutboxID int64 `gorm:"column:last_outbox_id"` + LastOccurredAt *time.Time `gorm:"column:last_occurred_at"` + ScannedRows int64 `gorm:"column:scanned_rows"` + AcceptedRows int64 `gorm:"column:accepted_rows"` + Completed bool `gorm:"column:completed"` + CreateTime time.Time `gorm:"column:create_time"` + UpdateTime time.Time `gorm:"column:update_time"` +} + +func (YumiGameKingBackfillState) TableName() string { return "yumi_game_king_backfill_state" } + // YumiGameKingUser 保存榜单聚合和已使用抽奖次数，避免排行榜扫描消费账本。 type YumiGameKingUser struct { ActivityID int64 `gorm:"column:activity_id;primaryKey;index:idx_yumi_gk_user_rank,priority:1"` diff --git a/internal/router/router.go b/internal/router/router.go index 506ff50..7eaa803 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -96,6 +96,7 @@ func NewRouter( registerRegisterRewardRoutes(engine, javaClient, services.RegisterReward) registerSignInRewardRoutes(engine, javaClient, services.SignInReward) registerTaskCenterRoutes(engine, cfg, javaClient, services.TaskCenter) + registerTaskCenterArchiveGCRoutes(engine, cfg, javaClient, services.TaskCenter) registerGameKingRoutes(engine, cfg, javaClient, services.GameKing) registerUserBadgeRoutes(engine, javaClient, services.UserBadge) registerVipRoutes(engine, javaClient, services.VIP) diff --git a/internal/router/task_center_archive_gc_routes.go b/internal/router/task_center_archive_gc_routes.go new file mode 100644 index 0000000..d5c2a42 --- /dev/null +++ b/internal/router/task_center_archive_gc_routes.go @@ -0,0 +1,81 @@ +package router + +import ( + "net/http" + + "chatapp3-golang/internal/config" + "chatapp3-golang/internal/service/taskcenter" + + "github.com/gin-gonic/gin" +) + +const taskCenterArchiveGCManagePermission = "resident-activity:task-center-archive-gc:manage" +const taskCenterArchiveGCMenuAlias = "ResidentTaskCenterArchiveGC" + +// registerTaskCenterArchiveGCRoutes 只做管理命令/cron tick 的 HTTP 绑定；所有状态机、 +// 租约、COS 校验和删除边界均由 taskcenter service 实现。 +func registerTaskCenterArchiveGCRoutes(engine *gin.Engine, cfg config.Config, javaClient authGateway, service *taskcenter.Service) { + if service == nil { + return + } + + admin := engine.Group("/resident-activity/task-center/archive-gc") + admin.Use(consoleAuthMiddleware(javaClient)) + admin.GET("/status", consoleMenuMiddleware(javaClient, taskCenterArchiveGCMenuAlias), func(c *gin.Context) { + response, err := service.ArchiveGCStatus(c.Request.Context()) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) + admin.POST("/dry-run", consolePermissionMiddleware(javaClient, taskCenterArchiveGCManagePermission), func(c *gin.Context) { + var request taskcenter.ArchiveGCDryRunRequest + if err := c.ShouldBindJSON(&request); err != nil { + writeError(c, taskcenter.NewAppError(http.StatusBadRequest, "bad_request", err.Error())) + return + } + response, err := service.ArchiveGCDryRun(c.Request.Context(), request) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) + admin.POST("/start", consolePermissionMiddleware(javaClient, taskCenterArchiveGCManagePermission), func(c *gin.Context) { + response, err := service.StartArchiveGC(c.Request.Context()) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) + admin.POST("/pause", consolePermissionMiddleware(javaClient, taskCenterArchiveGCManagePermission), func(c *gin.Context) { + response, err := service.PauseArchiveGC(c.Request.Context()) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) + admin.POST("/resume", consolePermissionMiddleware(javaClient, taskCenterArchiveGCManagePermission), func(c *gin.Context) { + response, err := service.ResumeArchiveGC(c.Request.Context()) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) + + internal := engine.Group("/internal/task-center/archive-gc") + // tick 会最终删除生产数据，密钥缺失也必须 fail-closed，不能沿用历史空密钥兼容。 + internal.Use(requiredInternalSecretMiddleware(cfg.HTTP.InternalCallbackSecret)) + internal.POST("/tick", func(c *gin.Context) { + response, err := service.ArchiveGCTick(c.Request.Context()) + if err != nil { + writeError(c, err) + return + } + writeOK(c, response) + }) +} diff --git a/internal/service/gameking/archive_safety.go b/internal/service/gameking/archive_safety.go new file mode 100644 index 0000000..045042e --- /dev/null +++ b/internal/service/gameking/archive_safety.go @@ -0,0 +1,65 @@ +package gameking + +import ( + "errors" + "net/http" + "strings" + + "chatapp3-golang/internal/model" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// ensureActivityEnableSafetyTx 把“索引已在线”和“GC 没在删除”纳入首次启用事务。 +// 它与 GC 删除事务锁同一控制行，因此两个并发请求不会同时各自通过旧状态检查。 +func ensureActivityEnableSafetyTx(tx *gorm.DB) error { + if err := requireBackfillIndexOnDB(tx); err != nil { + return err + } + var run model.TaskCenterArchiveGCRun + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", uint8(1)).First(&run).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_migration_required", "migration 060 archive GC safety state is required before enabling game king") + } + return NewAppError(http.StatusServiceUnavailable, "archive_gc_state_check_failed", "cannot verify archive GC state before enabling game king") + } + if strings.EqualFold(run.Status, "RUNNING") { + return NewAppError(http.StatusConflict, "archive_gc_running", "pause task center archive GC before enabling game king") + } + return nil +} + +func requireBackfillIndexOnDB(db *gorm.DB) error { + var columns []backfillIndexColumn + if err := db.Raw(` +SELECT COLUMN_NAME, SEQ_IN_INDEX, SUB_PART, INDEX_TYPE, IS_VISIBLE +FROM information_schema.STATISTICS +WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'task_center_event_archive_outbox' + AND INDEX_NAME = ? +ORDER BY SEQ_IN_INDEX ASC`, taskCenterGameKingBackfillIndex).Scan(&columns).Error; err != nil { + return NewAppError(http.StatusServiceUnavailable, "backfill_index_check_failed", "cannot verify the required task center backfill index") + } + + required := [...]string{"sys_origin", "event_type", "occurred_at", "id"} + if len(columns) == len(required) { + valid := true + for index, requiredName := range required { + column := columns[index] + if column.SeqInIndex != index+1 || !strings.EqualFold(column.ColumnName, requiredName) || column.SubPart != nil || + !strings.EqualFold(column.IndexType, "BTREE") || !strings.EqualFold(column.IsVisible, "YES") { + valid = false + break + } + } + if valid { + return nil + } + } + return NewAppError( + http.StatusServiceUnavailable, + "backfill_index_required", + "migration 058 backfill index (sys_origin,event_type,occurred_at,id) must be online before enabling or backfill", + ) +} diff --git a/internal/service/gameking/backfill_state.go b/internal/service/gameking/backfill_state.go new file mode 100644 index 0000000..87f4ecf --- /dev/null +++ b/internal/service/gameking/backfill_state.go @@ -0,0 +1,82 @@ +package gameking + +import ( + "context" + "net/http" + "time" + + "chatapp3-golang/internal/model" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// loadOrCreateBackfillState 先建立活动级水位行，后续并发补数才能通过行锁检测游标竞争， +// 避免较慢请求把较快请求已经推进的复合游标覆盖回旧位置。 +func (s *Service) loadOrCreateBackfillState(ctx context.Context, activity model.YumiGameKingActivity) (model.YumiGameKingBackfillState, error) { + now := time.Now() + start := activity.StartTime + initial := model.YumiGameKingBackfillState{ + ActivityID: activity.ID, LastOccurredAt: &start, CreateTime: now, UpdateTime: now, + } + if err := s.db.WithContext(ctx).Clauses(clause.OnConflict{DoNothing: true}).Create(&initial).Error; err != nil { + return initial, err + } + var row model.YumiGameKingBackfillState + if err := s.db.WithContext(ctx).Where("activity_id = ?", activity.ID).First(&row).Error; err != nil { + return row, err + } + return row, nil +} + +// adoptBackfillCursor 只兼容 migration 060 上线前调用方已经保存的合法 lastId；水位一旦 +// 非零，调用方就不能跳转或回退游标。 +func (s *Service) adoptBackfillCursor(ctx context.Context, activityID int64, cursor model.TaskCenterEventArchiveOutbox) (model.YumiGameKingBackfillState, error) { + var state model.YumiGameKingBackfillState + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("activity_id = ?", activityID).First(&state).Error; err != nil { + return err + } + if state.LastOutboxID != 0 { + return NewAppError(http.StatusConflict, "backfill_cursor_conflict", "persisted backfill cursor changed; reload and retry") + } + now := time.Now() + if err := tx.Model(&model.YumiGameKingBackfillState{}).Where("activity_id = ?", activityID).Updates(map[string]any{ + "last_outbox_id": cursor.ID, "last_occurred_at": cursor.OccurredAt, + "completed": false, "update_time": now, + }).Error; err != nil { + return err + } + state.LastOutboxID = cursor.ID + state.LastOccurredAt = &cursor.OccurredAt + state.UpdateTime = now + return nil + }) + return state, err +} + +func (s *Service) persistBackfillState( + ctx context.Context, + activityID, expectedLastID int64, + nextLastID int64, + nextOccurredAt time.Time, + scanned, accepted int, + completed bool, +) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var locked model.YumiGameKingBackfillState + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("activity_id = ?", activityID).First(&locked).Error; err != nil { + return err + } + if locked.LastOutboxID != expectedLastID { + // 本页可能已经通过幂等 ledger 入账；不覆盖赢家水位，调用方重读后重放即可安全收敛。 + return NewAppError(http.StatusConflict, "backfill_cursor_conflict", "another backfill request advanced the persisted cursor; reload and retry") + } + return tx.Model(&model.YumiGameKingBackfillState{}).Where("activity_id = ?", activityID).Updates(map[string]any{ + "last_outbox_id": nextLastID, "last_occurred_at": nextOccurredAt, + "scanned_rows": gorm.Expr("scanned_rows + ?", scanned), + "accepted_rows": gorm.Expr("accepted_rows + ?", accepted), + "completed": completed, "update_time": time.Now(), + }).Error + }) +} diff --git a/internal/service/gameking/config.go b/internal/service/gameking/config.go index 37f4d7b..e1751f5 100644 --- a/internal/service/gameking/config.go +++ b/internal/service/gameking/config.go @@ -175,6 +175,13 @@ func (s *Service) SaveActivity(ctx context.Context, req SaveActivityRequest) (*A if normalized.Enabled && !row.Enabled && !time.UnixMilli(normalized.StartTime).After(writeTime) { return NewAppError(http.StatusConflict, "activity_start_must_be_future", "the first enabled startTime must be in the future") } + // 已启用但尚未开始的活动仍可能修改时间窗；只要本次保存结果为 enabled， + // 就与 GC 删除事务锁同一控制行，避免 GC 按旧窗口保护后活动扩展到已删除区间。 + if normalized.Enabled { + if err := ensureActivityEnableSafetyTx(tx); err != nil { + return err + } + } if normalized.Enabled { if err := lockActivityScopeTx(tx, normalized.SysOrigin, writeTime); err != nil { return err @@ -275,6 +282,9 @@ func (s *Service) EnableActivity(ctx context.Context, activityID int64, enabled return NewAppError(http.StatusConflict, "activity_config_locked", "activity enable state cannot change after activity start") } if enabled { + if err := ensureActivityEnableSafetyTx(tx); err != nil { + return err + } if !locked.StartTime.After(lockedNow) { return NewAppError(http.StatusConflict, "activity_start_must_be_future", "the first enabled startTime must be in the future") } diff --git a/internal/service/gameking/event.go b/internal/service/gameking/event.go index 2f7011e..641dd57 100644 --- a/internal/service/gameking/event.go +++ b/internal/service/gameking/event.go @@ -175,8 +175,15 @@ func (s *Service) Backfill(ctx context.Context, activityID, lastID int64, limit if lastID < 0 { lastID = 0 } + state, err := s.loadOrCreateBackfillState(ctx, activity) + if err != nil { + return nil, err + } cursorTime := activity.StartTime - if lastID > 0 { + if state.LastOccurredAt != nil { + cursorTime = *state.LastOccurredAt + } + if lastID > 0 && state.LastOutboxID == 0 { var cursor model.TaskCenterEventArchiveOutbox if err := s.db.WithContext(ctx). Where("id = ? AND sys_origin = ? AND event_type = ? AND occurred_at >= ? AND occurred_at < ?", @@ -187,8 +194,16 @@ func (s *Service) Backfill(ctx context.Context, activityID, lastID int64, limit } return nil, err } + state, err = s.adoptBackfillCursor(ctx, activity.ID, cursor) + if err != nil { + return nil, err + } cursorTime = cursor.OccurredAt + } else if lastID > 0 && lastID != state.LastOutboxID { + return nil, NewAppError(http.StatusConflict, "backfill_cursor_conflict", "lastId does not match the persisted backfill cursor") } + lastID = state.LastOutboxID + startingLastID := lastID var rows []model.TaskCenterEventArchiveOutbox if err := s.db.WithContext(ctx). Where("sys_origin = ? AND event_type = ? AND occurred_at >= ? AND occurred_at < ? AND (occurred_at > ? OR (occurred_at = ? AND id > ?))", @@ -230,6 +245,10 @@ func (s *Service) Backfill(ctx context.Context, activityID, lastID int64, limit result.Accepted++ } result.NextLastID = row.ID + cursorTime = row.OccurredAt + } + if err := s.persistBackfillState(ctx, activity.ID, startingLastID, result.NextLastID, cursorTime, result.Scanned, result.Accepted, result.Finished); err != nil { + return nil, err } return result, nil } @@ -237,35 +256,5 @@ func (s *Service) Backfill(ctx context.Context, activityID, lastID int64, limit // requireBackfillIndex 只接受迁移 058 创建的完整 BTREE 列序；VARCHAR 前缀索引也会拒绝， // 因为它不能提供与完整等值前缀相同的执行计划保证。缺表、缺索引或列序漂移统一 fail-closed。 func (s *Service) requireBackfillIndex(ctx context.Context) error { - var columns []backfillIndexColumn - if err := s.db.WithContext(ctx).Raw(` -SELECT COLUMN_NAME, SEQ_IN_INDEX, SUB_PART, INDEX_TYPE, IS_VISIBLE -FROM information_schema.STATISTICS -WHERE TABLE_SCHEMA = DATABASE() - AND TABLE_NAME = 'task_center_event_archive_outbox' - AND INDEX_NAME = ? -ORDER BY SEQ_IN_INDEX ASC`, taskCenterGameKingBackfillIndex).Scan(&columns).Error; err != nil { - return NewAppError(http.StatusServiceUnavailable, "backfill_index_check_failed", "cannot verify the required task center backfill index") - } - - required := [...]string{"sys_origin", "event_type", "occurred_at", "id"} - if len(columns) == len(required) { - valid := true - for index, requiredName := range required { - column := columns[index] - if column.SeqInIndex != index+1 || !strings.EqualFold(column.ColumnName, requiredName) || column.SubPart != nil || - !strings.EqualFold(column.IndexType, "BTREE") || !strings.EqualFold(column.IsVisible, "YES") { - valid = false - break - } - } - if valid { - return nil - } - } - return NewAppError( - http.StatusServiceUnavailable, - "backfill_index_required", - "migration 058 backfill index (sys_origin,event_type,occurred_at,id) must be online before backfill", - ) + return requireBackfillIndexOnDB(s.db.WithContext(ctx)) } diff --git a/internal/service/taskcenter/archive.go b/internal/service/taskcenter/archive.go index 05e04cb..e86f958 100644 --- a/internal/service/taskcenter/archive.go +++ b/internal/service/taskcenter/archive.go @@ -215,7 +215,13 @@ func groupArchiveRows(rows []model.TaskCenterEventArchiveOutbox) [][]model.TaskC groupsByKey := make(map[string][]model.TaskCenterEventArchiveOutbox) keys := make([]string, 0) for _, row := range rows { - groupKey := row.SysOrigin + "\x00" + row.EventType + partitionHour := "UNKNOWN" + if !row.OccurredAt.IsZero() { + partitionHour = row.OccurredAt.UTC().Format("2006-01-02T15") + } + // COS key 带 UTC dt/hour 分区，所以同一抢占批次也必须先按小时拆组；否则跨小时 + // 事件会被放进第一行的错误目录，后续按分区恢复会漏数。 + groupKey := partitionHour + "\x00" + row.SysOrigin + "\x00" + row.EventType if _, exists := groupsByKey[groupKey]; !exists { keys = append(keys, groupKey) } @@ -238,15 +244,31 @@ func (s *Service) uploadArchiveGroup(ctx context.Context, client *cos.Client, ro if err != nil { return err } - _, err = client.Object.Put(ctx, cosKey, bytes.NewReader(body), &cos.ObjectPutOptions{ - ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{ - ContentType: "application/gzip", - ContentEncoding: "gzip", - }, + verified, err := putAndHeadArchiveObject(ctx, client, cosKey, body, map[string]string{ + "object-type": "archive", + "row-count": fmt.Sprintf("%d", len(rows)), }) if err != nil { return err } + verifiedAt := time.Now() + minID, maxID, minOccurredAt, maxOccurredAt := archiveGroupBounds(rows) + if err := s.upsertArchiveManifest(ctx, model.TaskCenterArchiveManifest{ + ObjectType: archiveManifestObjectArchive, + COSKey: cosKey, + SHA256: verified.SHA256, + ETag: verified.ETag, + CompressedSize: verified.CompressedSize, + RowCount: len(rows), + MinOutboxID: minID, + MaxOutboxID: maxID, + MinOccurredAt: minOccurredAt, + MaxOccurredAt: maxOccurredAt, + VerificationStatus: archiveManifestVerified, + VerifiedAt: &verifiedAt, + }); err != nil { + return err + } return s.markArchiveGroupUploaded(ctx, rows, cosKey) } diff --git a/internal/service/taskcenter/archive_gc_control.go b/internal/service/taskcenter/archive_gc_control.go new file mode 100644 index 0000000..0430515 --- /dev/null +++ b/internal/service/taskcenter/archive_gc_control.go @@ -0,0 +1,258 @@ +package taskcenter + +import ( + "context" + "errors" + "net/http" + "strconv" + "strings" + "time" + + "chatapp3-golang/internal/model" + "chatapp3-golang/internal/utils" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +const ( + archiveGCControlID = uint8(1) + defaultArchiveGCBatch = 200 + minArchiveGCBatch = 200 + maxArchiveGCBatch = 500 + minimumArchiveGCAge = 30 * 24 * time.Hour + archiveGCClaimIndex = "idx_task_center_archive_claim_create" +) + +// ArchiveGCStatus 返回唯一控制行；总开关关闭时不要求迁移已经执行，便于新版本先安全上线。 +func (s *Service) ArchiveGCStatus(ctx context.Context) (*ArchiveGCStatusView, error) { + row, err := s.loadArchiveGCRun(ctx) + if err != nil { + // 总开关关闭的新环境可能尚未执行 060；此时返回安全的 IDLE。若控制行已经存在， + // 即使关闭开关也展示真实 RUNNING/PAUSED 状态，避免运维误以为历史 run 不存在。 + if !s.cfg.TaskCenter.Archive.GC.Enabled { + return &ArchiveGCStatusView{Enabled: false, Status: ArchiveGCStatusIdle, BatchSize: defaultArchiveGCBatch}, nil + } + return nil, err + } + view := s.archiveGCStatusView(row) + return &view, nil +} + +// ArchiveGCDryRun 对索引支持的候选范围做有界预检，并把 cutoff 固定进控制行。 +// 这里不做 COUNT(*)，因为在大表上为了展示总数触发长扫描本身就违背清理降载目标。 +func (s *Service) ArchiveGCDryRun(ctx context.Context, request ArchiveGCDryRunRequest) (*ArchiveGCDryRunResponse, error) { + if err := s.requireArchiveGCReady(ctx); err != nil { + return nil, err + } + current, err := s.loadArchiveGCRun(ctx) + if err != nil { + return nil, err + } + if current.Status == ArchiveGCStatusRunning { + return nil, NewAppError(http.StatusConflict, "archive_gc_running", "pause the running archive GC before preparing another cutoff") + } + cutoff := time.UnixMilli(request.CutoffTime) + if request.CutoffTime <= 0 || cutoff.After(time.Now().Add(-minimumArchiveGCAge)) { + return nil, NewAppError(http.StatusBadRequest, "archive_gc_cutoff_too_recent", "cutoffTime must retain at least 30 days of hot data") + } + batchSize, err := normalizeArchiveGCBatch(request.BatchSize) + if err != nil { + return nil, err + } + preview, err := s.findArchiveGCCandidates(ctx, cutoff, batchSize+1) + if err != nil { + return nil, err + } + hasMore := len(preview) > batchSize + if hasMore { + preview = preview[:batchSize] + } + runSnowflake, err := utils.NextID() + if err != nil { + return nil, err + } + runID := strconv.FormatInt(runSnowflake, 10) + if runID == "" { + return nil, NewAppError(http.StatusInternalServerError, "archive_gc_run_id_failed", "cannot create archive GC run id") + } + now := time.Now() + var updated model.TaskCenterArchiveGCRun + err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var locked model.TaskCenterArchiveGCRun + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&locked).Error; err != nil { + return archiveGCMigrationError(err) + } + if locked.Status == ArchiveGCStatusRunning { + return NewAppError(http.StatusConflict, "archive_gc_running", "pause the running archive GC before preparing another cutoff") + } + updates := map[string]any{ + "run_id": runID, "status": ArchiveGCStatusDryRun, "cutoff_time": cutoff, "batch_size": batchSize, + "scanned_rows": 0, "eligible_rows": 0, "protected_rows": 0, "deleted_rows": 0, + "recovery_object_count": 0, "cursor_create_time": nil, "cursor_id": 0, + "lease_owner": "", "lease_until": nil, "last_error": "", "update_time": now, + } + if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(updates).Error; err != nil { + return err + } + return tx.Where("id = ?", archiveGCControlID).First(&updated).Error + }) + if err != nil { + return nil, err + } + view := s.archiveGCStatusView(updated) + return &ArchiveGCDryRunResponse{ArchiveGCStatusView: view, PreviewRows: len(preview), HasMore: hasMore}, nil +} + +// StartArchiveGC 只允许启动刚刚预检过的控制行，确保真实删除沿用完全相同的固定 cutoff。 +func (s *Service) StartArchiveGC(ctx context.Context) (*ArchiveGCStatusView, error) { + if err := s.requireArchiveGCReady(ctx); err != nil { + return nil, err + } + return s.transitionArchiveGC(ctx, []string{ArchiveGCStatusDryRun}, ArchiveGCStatusRunning) +} + +func (s *Service) PauseArchiveGC(ctx context.Context) (*ArchiveGCStatusView, error) { + if err := s.requireArchiveGCEnabled(); err != nil { + return nil, err + } + return s.transitionArchiveGC(ctx, []string{ArchiveGCStatusRunning}, ArchiveGCStatusPaused) +} + +func (s *Service) ResumeArchiveGC(ctx context.Context) (*ArchiveGCStatusView, error) { + if err := s.requireArchiveGCReady(ctx); err != nil { + return nil, err + } + return s.transitionArchiveGC(ctx, []string{ArchiveGCStatusPaused, ArchiveGCStatusFailed}, ArchiveGCStatusRunning) +} + +func (s *Service) transitionArchiveGC(ctx context.Context, from []string, to string) (*ArchiveGCStatusView, error) { + var updated model.TaskCenterArchiveGCRun + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var locked model.TaskCenterArchiveGCRun + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&locked).Error; err != nil { + return archiveGCMigrationError(err) + } + allowed := false + for _, status := range from { + if locked.Status == status { + allowed = true + break + } + } + if !allowed { + return NewAppError(http.StatusConflict, "archive_gc_state_conflict", "archive GC state does not allow this operation") + } + if locked.CutoffTime == nil { + return NewAppError(http.StatusConflict, "archive_gc_cutoff_missing", "run dry-run before starting archive GC") + } + now := time.Now() + if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(map[string]any{ + "status": to, "lease_owner": "", "lease_until": nil, "last_error": "", "update_time": now, + }).Error; err != nil { + return err + } + return tx.Where("id = ?", archiveGCControlID).First(&updated).Error + }) + if err != nil { + return nil, err + } + view := s.archiveGCStatusView(updated) + return &view, nil +} + +func (s *Service) loadArchiveGCRun(ctx context.Context) (model.TaskCenterArchiveGCRun, error) { + var row model.TaskCenterArchiveGCRun + if err := s.db.WithContext(ctx).Where("id = ?", archiveGCControlID).First(&row).Error; err != nil { + return row, archiveGCMigrationError(err) + } + return row, nil +} + +func (s *Service) requireArchiveGCReady(ctx context.Context) error { + if err := s.requireArchiveGCEnabled(); err != nil { + return err + } + archive := s.cfg.TaskCenter.Archive + if strings.TrimSpace(archive.Bucket) == "" || strings.TrimSpace(archive.Region) == "" || + strings.TrimSpace(archive.SecretID) == "" || strings.TrimSpace(archive.SecretKey) == "" { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_cos_not_configured", "task center archive COS is not configured") + } + return s.requireArchiveGCClaimIndex(ctx) +} + +func (s *Service) requireArchiveGCEnabled() error { + if !s.cfg.TaskCenter.Archive.GC.Enabled { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_disabled", "task center archive GC is disabled") + } + return nil +} + +func (s *Service) requireArchiveGCClaimIndex(ctx context.Context) error { + type indexColumn struct { + ColumnName string `gorm:"column:column_name"` + Seq int `gorm:"column:seq_in_index"` + SubPart *int64 `gorm:"column:sub_part"` + IndexType string `gorm:"column:index_type"` + IsVisible string `gorm:"column:is_visible"` + } + var columns []indexColumn + if err := s.db.WithContext(ctx).Raw(` +SELECT COLUMN_NAME, SEQ_IN_INDEX, SUB_PART, INDEX_TYPE, IS_VISIBLE +FROM information_schema.STATISTICS +WHERE TABLE_SCHEMA = DATABASE() + AND TABLE_NAME = 'task_center_event_archive_outbox' + AND INDEX_NAME = ? +ORDER BY SEQ_IN_INDEX`, archiveGCClaimIndex).Scan(&columns).Error; err != nil { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_index_check_failed", "cannot verify the archive GC claim index") + } + required := [...]string{"status", "create_time", "id"} + if len(columns) != len(required) { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_index_required", "migration 056 claim index must be online before archive GC") + } + for index, expected := range required { + column := columns[index] + if column.Seq != index+1 || !strings.EqualFold(column.ColumnName, expected) || column.SubPart != nil || + !strings.EqualFold(column.IndexType, "BTREE") || !strings.EqualFold(column.IsVisible, "YES") { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_index_required", "migration 056 claim index must be online before archive GC") + } + } + return nil +} + +func normalizeArchiveGCBatch(value int) (int, error) { + if value == 0 { + return defaultArchiveGCBatch, nil + } + if value < minArchiveGCBatch || value > maxArchiveGCBatch { + return 0, NewAppError(http.StatusBadRequest, "archive_gc_batch_invalid", "batchSize must be between 200 and 500") + } + return value, nil +} + +func archiveGCMigrationError(err error) error { + if errors.Is(err, gorm.ErrRecordNotFound) { + return NewAppError(http.StatusServiceUnavailable, "archive_gc_migration_required", "migration 060 archive GC control row is required") + } + return err +} + +func (s *Service) archiveGCStatusView(row model.TaskCenterArchiveGCRun) ArchiveGCStatusView { + view := ArchiveGCStatusView{ + Enabled: s.cfg.TaskCenter.Archive.GC.Enabled, RunID: row.RunID, Status: row.Status, + BatchSize: row.BatchSize, ScannedRows: row.ScannedRows, EligibleRows: row.EligibleRows, + ProtectedRows: row.ProtectedRows, DeletedRows: row.DeletedRows, + RecoveryObjectCount: row.RecoveryObjectCount, CursorID: row.CursorID, + LastError: row.LastError, CreateTime: row.CreateTime.UnixMilli(), UpdateTime: row.UpdateTime.UnixMilli(), + } + if row.CutoffTime != nil { + view.CutoffTime = row.CutoffTime.UnixMilli() + } + if row.CursorCreateTime != nil { + view.CursorCreateTime = row.CursorCreateTime.UnixMilli() + } + if row.LeaseUntil != nil { + view.LeaseUntil = row.LeaseUntil.UnixMilli() + } + return view +} diff --git a/internal/service/taskcenter/archive_gc_types.go b/internal/service/taskcenter/archive_gc_types.go new file mode 100644 index 0000000..1af4eaa --- /dev/null +++ b/internal/service/taskcenter/archive_gc_types.go @@ -0,0 +1,53 @@ +package taskcenter + +const ( + ArchiveGCStatusIdle = "IDLE" + ArchiveGCStatusDryRun = "DRY_RUN" + ArchiveGCStatusRunning = "RUNNING" + ArchiveGCStatusPaused = "PAUSED" + ArchiveGCStatusCompleted = "COMPLETED" + ArchiveGCStatusFailed = "FAILED" +) + +// ArchiveGCDryRunRequest 冻结一次清理的截止时间和批量；start 不再接收这些字段，避免 +// 预检与真实执行使用不同范围。 +type ArchiveGCDryRunRequest struct { + CutoffTime int64 `json:"cutoffTime"` + BatchSize int `json:"batchSize"` +} + +type ArchiveGCStatusView struct { + Enabled bool `json:"enabled"` + RunID string `json:"runId"` + Status string `json:"status"` + CutoffTime int64 `json:"cutoffTime,omitempty"` + BatchSize int `json:"batchSize"` + ScannedRows int64 `json:"scannedRows"` + EligibleRows int64 `json:"eligibleRows"` + ProtectedRows int64 `json:"protectedRows"` + DeletedRows int64 `json:"deletedRows"` + RecoveryObjectCount int64 `json:"recoveryObjectCount"` + CursorCreateTime int64 `json:"cursorCreateTime,omitempty"` + CursorID int64 `json:"cursorId,string,omitempty"` + LeaseUntil int64 `json:"leaseUntil,omitempty"` + LastError string `json:"lastError,omitempty"` + CreateTime int64 `json:"createTime,omitempty"` + UpdateTime int64 `json:"updateTime,omitempty"` +} + +type ArchiveGCDryRunResponse struct { + ArchiveGCStatusView + PreviewRows int `json:"previewRows"` + HasMore bool `json:"hasMore"` +} + +type ArchiveGCTickResponse struct { + RunID string `json:"runId"` + Status string `json:"status"` + ProcessedRows int `json:"processedRows"` + DeletedRows int `json:"deletedRows"` + ProtectedRows int `json:"protectedRows"` + RecoveryCOSKey string `json:"recoveryCosKey,omitempty"` + Finished bool `json:"finished"` + SkippedReason string `json:"skippedReason,omitempty"` +} diff --git a/internal/service/taskcenter/archive_gc_work.go b/internal/service/taskcenter/archive_gc_work.go new file mode 100644 index 0000000..85e1daf --- /dev/null +++ b/internal/service/taskcenter/archive_gc_work.go @@ -0,0 +1,533 @@ +package taskcenter + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "path" + "sort" + "strconv" + "strings" + "time" + + "chatapp3-golang/internal/model" + + "github.com/tencentyun/cos-go-sdk-v5" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +const ( + archiveGCLeaseDuration = 5 * time.Minute + maxArchiveGCSourceObjectsPerRun = 10 +) + +var archiveGCWorkerIdentity = buildArchiveGCWorkerIdentity() + +type archiveGCRecoveryRow struct { + ID int64 `json:"id,string"` + SysOrigin string `json:"sysOrigin"` + EventID string `json:"eventId"` + EventType string `json:"eventType"` + UserID int64 `json:"userId,string"` + OccurredAt time.Time `json:"occurredAt"` + RawJSON string `json:"rawJson"` + Status string `json:"status"` + RetryCount int `json:"retryCount"` + COSKey string `json:"cosKey"` + FailureReason string `json:"failureReason"` + CreateTime time.Time `json:"createTime"` + UpdateTime time.Time `json:"updateTime"` + ArchivedAt *time.Time `json:"archivedAt"` +} + +// ArchiveGCTick 由 chatapp-cron 显式触发且每次只处理一个小批次。COS 回读和恢复包上传都 +// 位于数据库删除之前；任一步失败都会暂停本次 run，绝不继续“尽力删除”。 +func (s *Service) ArchiveGCTick(ctx context.Context) (*ArchiveGCTickResponse, error) { + if err := s.requireArchiveGCReady(ctx); err != nil { + return nil, err + } + leaseToken := fmt.Sprintf("%s:%d", archiveGCWorkerIdentity, time.Now().UnixNano()) + run, acquired, reason, err := s.acquireArchiveGCLease(ctx, leaseToken) + if err != nil { + return nil, err + } + if !acquired { + return &ArchiveGCTickResponse{RunID: run.RunID, Status: run.Status, SkippedReason: reason, Finished: run.Status == ArchiveGCStatusCompleted}, nil + } + if run.CutoffTime == nil { + err := NewAppError(http.StatusConflict, "archive_gc_cutoff_missing", "archive GC cutoff is missing") + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + + rows, err := s.findArchiveGCCandidates(ctx, *run.CutoffTime, run.BatchSize) + if err != nil { + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + if len(rows) == 0 { + if err := s.completeArchiveGCTick(ctx, run.RunID, leaseToken); err != nil { + return nil, err + } + return &ArchiveGCTickResponse{RunID: run.RunID, Status: ArchiveGCStatusCompleted, Finished: true}, nil + } + rows = limitArchiveGCSourceObjects(rows, maxArchiveGCSourceObjectsPerRun) + client, err := s.newCOSClient() + if err != nil { + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + if err := s.verifyArchiveGCSources(ctx, client, rows); err != nil { + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + recoveryKey, recoverySHA, err := s.createAndVerifyArchiveGCRecovery(ctx, client, run.RunID, rows) + if err != nil { + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + deleted, protected, err := s.deleteVerifiedArchiveGCRows(ctx, run, leaseToken, recoveryKey, recoverySHA, rows) + if err != nil { + s.pauseArchiveGCTick(run.RunID, leaseToken, err) + return nil, err + } + return &ArchiveGCTickResponse{ + RunID: run.RunID, Status: ArchiveGCStatusRunning, ProcessedRows: len(rows), DeletedRows: deleted, + ProtectedRows: protected, RecoveryCOSKey: recoveryKey, + }, nil +} + +func (s *Service) findArchiveGCCandidates(ctx context.Context, cutoff time.Time, limit int) ([]model.TaskCenterEventArchiveOutbox, error) { + if limit <= 0 || limit > maxArchiveGCBatch+1 { + limit = defaultArchiveGCBatch + } + var rows []model.TaskCenterEventArchiveOutbox + // FORCE INDEX 将扫描固定在 (status,create_time,id)；NOT EXISTS 只针对命中的小范围行 + // 查询活动小表。MAX_EXECUTION_TIME 防止长期活动保护了大量最旧行时，为凑满一批而 + // 在生产大表持续扫描；超时会让 tick 暂停且不产生 DELETE。 + err := s.db.WithContext(ctx).Raw(` +SELECT /*+ MAX_EXECUTION_TIME(5000) */ archive_row.* +FROM task_center_event_archive_outbox AS archive_row +FORCE INDEX (idx_task_center_archive_claim_create) +WHERE archive_row.status = ? + AND archive_row.create_time < ? + AND archive_row.archived_at IS NOT NULL + AND archive_row.archived_at < ? + AND archive_row.occurred_at IS NOT NULL + AND archive_row.occurred_at < ? + AND archive_row.cos_key <> '' + AND ( + archive_row.event_type <> ? + OR NOT EXISTS ( + SELECT 1 + FROM yumi_game_king_activity AS activity + WHERE activity.enabled = 1 + AND activity.settlement_status <> 'COMPLETED' + AND activity.sys_origin = archive_row.sys_origin + AND archive_row.occurred_at >= activity.start_time + AND archive_row.occurred_at < activity.end_time + ) + ) +ORDER BY archive_row.create_time ASC, archive_row.id ASC +LIMIT ?`, ArchiveStatusUploaded, cutoff, cutoff, cutoff, EventTypeGameConsumeGold, limit).Scan(&rows).Error + return rows, err +} + +func (s *Service) acquireArchiveGCLease(ctx context.Context, leaseToken string) (model.TaskCenterArchiveGCRun, bool, string, error) { + var run model.TaskCenterArchiveGCRun + claimed := false + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&run).Error; err != nil { + return archiveGCMigrationError(err) + } + if run.Status != ArchiveGCStatusRunning { + return nil + } + var databaseClock struct { + Now time.Time `gorm:"column:now_time"` + } + if err := tx.Raw("SELECT NOW(3) AS now_time").Scan(&databaseClock).Error; err != nil { + return err + } + now := databaseClock.Now + if now.IsZero() { + return fmt.Errorf("database clock returned zero time") + } + if run.LeaseUntil != nil && run.LeaseUntil.After(now) { + return nil + } + leaseUntil := now.Add(archiveGCLeaseDuration) + if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(map[string]any{ + "lease_owner": leaseToken, "lease_until": leaseUntil, "update_time": now, + }).Error; err != nil { + return err + } + run.LeaseOwner = leaseToken + run.LeaseUntil = &leaseUntil + claimed = true + return nil + }) + if err != nil { + return run, false, "", err + } + if run.Status != ArchiveGCStatusRunning { + return run, false, "not_running", nil + } + if !claimed { + return run, false, "lease_busy", nil + } + return run, true, "", nil +} + +func (s *Service) verifyArchiveGCSources(ctx context.Context, client *cos.Client, rows []model.TaskCenterEventArchiveOutbox) error { + groups := make(map[string][]model.TaskCenterEventArchiveOutbox) + for _, row := range rows { + groups[row.COSKey] = append(groups[row.COSKey], row) + } + keys := make([]string, 0, len(groups)) + for key := range groups { + keys = append(keys, key) + } + sort.Strings(keys) + for _, cosKey := range keys { + // HEAD 确认对象当前版本，GET+gunzip 再确认候选 raw_json 确实包含在对象内； + // 不能因为数据库已有 cos_key 就假设历史 PUT 完整成功。 + head, err := headAndVerifyArchiveObject(ctx, client, cosKey, "", -1, false) + if err != nil { + return fmt.Errorf("verify source archive %s HEAD: %w", cosKey, err) + } + if head.CompressedSize > maxArchiveCOSObjectBytes { + return fmt.Errorf("source archive %s exceeds %d bytes", cosKey, maxArchiveCOSObjectBytes) + } + body, _, err := downloadArchiveObject(ctx, client, cosKey) + if err != nil { + return fmt.Errorf("verify source archive %s GET: %w", cosKey, err) + } + if int64(len(body)) != head.CompressedSize { + return fmt.Errorf("source archive %s changed between HEAD and GET", cosKey) + } + sha := archiveSHA256(body) + verifiedHead, err := headAndVerifyArchiveObject(ctx, client, cosKey, sha, int64(len(body)), false) + if err != nil { + return fmt.Errorf("verify source archive %s metadata: %w", cosKey, err) + } + lines, err := archiveObjectLines(body) + if err != nil { + return fmt.Errorf("verify source archive %s gzip: %w", cosKey, err) + } + available := make(map[string]int, len(lines)) + for _, line := range lines { + available[archiveLineSHA256(line)]++ + } + for _, row := range groups[cosKey] { + hash := archiveLineSHA256(row.RawJSON) + if available[hash] <= 0 { + return fmt.Errorf("source archive %s does not contain outbox row %d", cosKey, row.ID) + } + available[hash]-- + } + var existing model.TaskCenterArchiveManifest + existingErr := s.db.WithContext(ctx).Where("cos_key = ?", cosKey).First(&existing).Error + if existingErr == nil && existing.SHA256 != "" && !strings.EqualFold(existing.SHA256, sha) { + return fmt.Errorf("source archive %s no longer matches its manifest", cosKey) + } + if existingErr != nil && !errors.Is(existingErr, gorm.ErrRecordNotFound) { + return existingErr + } + verifiedAt := time.Now() + if err := s.upsertArchiveManifest(ctx, model.TaskCenterArchiveManifest{ + ObjectType: archiveManifestObjectArchive, COSKey: cosKey, SHA256: sha, ETag: verifiedHead.ETag, + CompressedSize: int64(len(body)), RowCount: len(lines), MinOutboxID: existing.MinOutboxID, + MaxOutboxID: existing.MaxOutboxID, MinOccurredAt: existing.MinOccurredAt, + MaxOccurredAt: existing.MaxOccurredAt, VerificationStatus: archiveManifestVerified, + VerifiedAt: &verifiedAt, + }); err != nil { + return err + } + } + return nil +} + +func (s *Service) createAndVerifyArchiveGCRecovery( + ctx context.Context, + client *cos.Client, + runID string, + rows []model.TaskCenterEventArchiveOutbox, +) (string, string, error) { + body, err := gzipArchiveGCRecoveryRows(rows) + if err != nil { + return "", "", err + } + cosKey := s.buildArchiveGCRecoveryKey(runID, rows) + verified, err := putAndHeadArchiveObject(ctx, client, cosKey, body, map[string]string{ + "object-type": "gc-recovery", "run-id": runID, "row-count": strconv.Itoa(len(rows)), + }) + if err != nil { + return "", "", err + } + // 恢复包本身也必须 GET+gunzip 并逐行比对 ID/raw_json，HEAD 成功不等于内容可恢复。 + downloaded, _, err := downloadArchiveObject(ctx, client, cosKey) + if err != nil { + return "", "", err + } + if archiveSHA256(downloaded) != verified.SHA256 { + return "", "", fmt.Errorf("recovery object %s sha256 mismatch after upload", cosKey) + } + lines, err := archiveObjectLines(downloaded) + if err != nil { + return "", "", err + } + if len(lines) != len(rows) { + return "", "", fmt.Errorf("recovery object %s row count %d does not match %d", cosKey, len(lines), len(rows)) + } + expected := make(map[int64]string, len(rows)) + for _, row := range rows { + expected[row.ID] = archiveLineSHA256(row.RawJSON) + } + for _, line := range lines { + var recovered archiveGCRecoveryRow + if err := json.Unmarshal([]byte(line), &recovered); err != nil { + return "", "", fmt.Errorf("decode recovery object %s: %w", cosKey, err) + } + expectedHash, ok := expected[recovered.ID] + if !ok || expectedHash != archiveLineSHA256(recovered.RawJSON) { + return "", "", fmt.Errorf("recovery object %s has unexpected row %d", cosKey, recovered.ID) + } + delete(expected, recovered.ID) + } + if len(expected) != 0 { + return "", "", fmt.Errorf("recovery object %s is missing rows", cosKey) + } + verifiedAt := time.Now() + minID, maxID, minOccurredAt, maxOccurredAt := archiveGroupBounds(rows) + parentCOSKey := "" + if rowsShareArchiveCOSKey(rows) { + parentCOSKey = rows[0].COSKey + } + if err := s.upsertArchiveManifest(ctx, model.TaskCenterArchiveManifest{ + ObjectType: archiveManifestObjectRecovery, RunID: runID, COSKey: cosKey, ParentCOSKey: parentCOSKey, + SHA256: verified.SHA256, ETag: verified.ETag, CompressedSize: verified.CompressedSize, + RowCount: len(rows), MinOutboxID: minID, MaxOutboxID: maxID, MinOccurredAt: minOccurredAt, + MaxOccurredAt: maxOccurredAt, VerificationStatus: archiveManifestVerified, VerifiedAt: &verifiedAt, + }); err != nil { + return "", "", err + } + return cosKey, verified.SHA256, nil +} + +func (s *Service) deleteVerifiedArchiveGCRows( + ctx context.Context, + run model.TaskCenterArchiveGCRun, + leaseToken, recoveryKey, recoverySHA string, + candidates []model.TaskCenterEventArchiveOutbox, +) (int, int, error) { + if run.CutoffTime == nil { + return 0, 0, NewAppError(http.StatusConflict, "archive_gc_cutoff_missing", "archive GC cutoff is missing") + } + candidateByID := make(map[int64]model.TaskCenterEventArchiveOutbox, len(candidates)) + ids := make([]int64, 0, len(candidates)) + for _, row := range candidates { + candidateByID[row.ID] = row + ids = append(ids, row.ID) + } + deleted, protected := 0, 0 + err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var lockedRun model.TaskCenterArchiveGCRun + // 与 Game King 启用门禁锁同一行。GC 在这把锁下复核活动窗口并删行，启用请求 + // 要么先完成并被本事务看到，要么等待删除提交后发现 GC 仍为 RUNNING 而失败。 + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&lockedRun).Error; err != nil { + return archiveGCMigrationError(err) + } + if lockedRun.RunID != run.RunID || lockedRun.Status != ArchiveGCStatusRunning || lockedRun.LeaseOwner != leaseToken { + return NewAppError(http.StatusConflict, "archive_gc_lease_lost", "archive GC lease or state changed before delete") + } + var recoveryManifest model.TaskCenterArchiveManifest + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where( + "cos_key = ? AND run_id = ? AND object_type = ? AND verification_status = ? AND sha256 = ?", + recoveryKey, run.RunID, archiveManifestObjectRecovery, archiveManifestVerified, recoverySHA, + ).First(&recoveryManifest).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return NewAppError(http.StatusConflict, "archive_gc_recovery_manifest_missing", "verified recovery manifest changed or is missing before delete") + } + return err + } + if recoveryManifest.RowCount != len(candidates) || recoveryManifest.VerifiedAt == nil { + return NewAppError(http.StatusConflict, "archive_gc_recovery_manifest_invalid", "recovery manifest row count or verification time is invalid") + } + + var lockedRows []model.TaskCenterEventArchiveOutbox + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id IN ?", ids).Order("create_time ASC, id ASC").Find(&lockedRows).Error; err != nil { + return err + } + var protectedActivities []model.YumiGameKingActivity + if err := tx.Where("enabled = ? AND settlement_status <> ?", true, "COMPLETED").Find(&protectedActivities).Error; err != nil { + return err + } + deleteRows := make([]model.TaskCenterEventArchiveOutbox, 0, len(lockedRows)) + for _, row := range lockedRows { + original, exists := candidateByID[row.ID] + if !exists || row.Status != ArchiveStatusUploaded || row.COSKey == "" || row.COSKey != original.COSKey || + archiveLineSHA256(row.RawJSON) != archiveLineSHA256(original.RawJSON) || row.ArchivedAt == nil || + !row.CreateTime.Before(*run.CutoffTime) || !row.ArchivedAt.Before(*run.CutoffTime) || + row.OccurredAt.IsZero() || !row.OccurredAt.Before(*run.CutoffTime) || archiveGCActivityProtects(row, protectedActivities) { + protected++ + continue + } + deleteRows = append(deleteRows, row) + } + if missing := len(candidates) - len(lockedRows); missing > 0 { + // 另一个写路径删除/改动候选属于异常状态；恢复包仍完整，但本次只统计为保护跳过。 + protected += missing + } + deleteIDs := archiveRowIDs(deleteRows) + if len(deleteIDs) > 0 { + result := tx.Where("id IN ? AND status = ?", deleteIDs, ArchiveStatusUploaded).Delete(&model.TaskCenterEventArchiveOutbox{}) + if result.Error != nil { + return result.Error + } + deleted = int(result.RowsAffected) + if deleted != len(deleteIDs) { + return fmt.Errorf("archive GC deleted %d rows, expected %d", deleted, len(deleteIDs)) + } + } + now := time.Now() + updates := map[string]any{ + "scanned_rows": gorm.Expr("scanned_rows + ?", len(candidates)), + "eligible_rows": gorm.Expr("eligible_rows + ?", len(deleteRows)), + "protected_rows": gorm.Expr("protected_rows + ?", protected), + "deleted_rows": gorm.Expr("deleted_rows + ?", deleted), + "recovery_object_count": gorm.Expr("recovery_object_count + 1"), + "lease_owner": "", "lease_until": nil, "last_error": "", "update_time": now, + } + if deleted > 0 { + last := deleteRows[len(deleteRows)-1] + updates["cursor_create_time"] = last.CreateTime + updates["cursor_id"] = last.ID + } + if err := tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(updates).Error; err != nil { + return err + } + return nil + }) + return deleted, protected, err +} + +func (s *Service) completeArchiveGCTick(ctx context.Context, runID, leaseToken string) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var locked model.TaskCenterArchiveGCRun + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", archiveGCControlID).First(&locked).Error; err != nil { + return archiveGCMigrationError(err) + } + if locked.RunID != runID || locked.Status != ArchiveGCStatusRunning || locked.LeaseOwner != leaseToken { + return NewAppError(http.StatusConflict, "archive_gc_lease_lost", "archive GC lease or state changed before completion") + } + return tx.Model(&model.TaskCenterArchiveGCRun{}).Where("id = ?", archiveGCControlID).Updates(map[string]any{ + "status": ArchiveGCStatusCompleted, "lease_owner": "", "lease_until": nil, + "last_error": "", "update_time": time.Now(), + }).Error + }) +} + +func (s *Service) pauseArchiveGCTick(runID, leaseToken string, workErr error) { + // HTTP/cron 超时往往已取消传入 ctx；失败状态必须用独立短上下文落库，否则控制行会 + // 继续显示 RUNNING，直到租约过期后再次触发同一批 COS 工作。 + failureCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + message := "archive GC tick failed" + if workErr != nil { + message = truncateTaskCenterFailure(workErr.Error()) + } + _ = s.db.WithContext(failureCtx).Model(&model.TaskCenterArchiveGCRun{}). + Where("id = ? AND run_id = ? AND status = ? AND lease_owner = ?", archiveGCControlID, runID, ArchiveGCStatusRunning, leaseToken). + Updates(map[string]any{ + "status": ArchiveGCStatusPaused, "lease_owner": "", "lease_until": nil, + "last_error": message, "update_time": time.Now(), + }).Error +} + +func gzipArchiveGCRecoveryRows(rows []model.TaskCenterEventArchiveOutbox) ([]byte, error) { + var buffer bytes.Buffer + writer := gzip.NewWriter(&buffer) + encoder := json.NewEncoder(writer) + for _, row := range rows { + if err := encoder.Encode(archiveGCRecoveryRow{ + ID: row.ID, SysOrigin: row.SysOrigin, EventID: row.EventID, EventType: row.EventType, + UserID: row.UserID, OccurredAt: row.OccurredAt, RawJSON: row.RawJSON, Status: row.Status, + RetryCount: row.RetryCount, COSKey: row.COSKey, FailureReason: row.FailureReason, + CreateTime: row.CreateTime, UpdateTime: row.UpdateTime, ArchivedAt: row.ArchivedAt, + }); err != nil { + _ = writer.Close() + return nil, err + } + } + if err := writer.Close(); err != nil { + return nil, err + } + return buffer.Bytes(), nil +} + +func (s *Service) buildArchiveGCRecoveryKey(runID string, rows []model.TaskCenterEventArchiveOutbox) string { + prefix := strings.Trim(strings.TrimSpace(s.cfg.TaskCenter.Archive.Prefix), "/") + if prefix == "" { + prefix = "task-center-events/v1" + } + return path.Join(prefix, "gc-recovery", "run="+sanitizeArchivePartition(runID), fmt.Sprintf( + "part-%s-%d-%d.jsonl.gz", time.Now().UTC().Format("20060102T150405.000000000Z"), rows[0].ID, len(rows), + )) +} + +func archiveGCActivityProtects(row model.TaskCenterEventArchiveOutbox, activities []model.YumiGameKingActivity) bool { + if row.EventType != EventTypeGameConsumeGold { + return false + } + for _, activity := range activities { + if activity.SysOrigin == row.SysOrigin && !row.OccurredAt.Before(activity.StartTime) && row.OccurredAt.Before(activity.EndTime) { + return true + } + } + return false +} + +func limitArchiveGCSourceObjects(rows []model.TaskCenterEventArchiveOutbox, maximum int) []model.TaskCenterEventArchiveOutbox { + if maximum <= 0 { + return nil + } + seen := make(map[string]struct{}, maximum) + for index, row := range rows { + if _, exists := seen[row.COSKey]; exists { + continue + } + if len(seen) == maximum { + return rows[:index] + } + seen[row.COSKey] = struct{}{} + } + return rows +} + +func rowsShareArchiveCOSKey(rows []model.TaskCenterEventArchiveOutbox) bool { + if len(rows) == 0 { + return false + } + key := rows[0].COSKey + for _, row := range rows[1:] { + if row.COSKey != key { + return false + } + } + return true +} + +func buildArchiveGCWorkerIdentity() string { + hostname, _ := os.Hostname() + if strings.TrimSpace(hostname) == "" { + hostname = "unknown-host" + } + return fmt.Sprintf("%s:%d", hostname, os.Getpid()) +} diff --git a/internal/service/taskcenter/archive_manifest.go b/internal/service/taskcenter/archive_manifest.go new file mode 100644 index 0000000..94fa6ad --- /dev/null +++ b/internal/service/taskcenter/archive_manifest.go @@ -0,0 +1,74 @@ +package taskcenter + +import ( + "context" + "time" + + "chatapp3-golang/internal/model" + + "gorm.io/gorm/clause" +) + +const ( + archiveManifestObjectArchive = "ARCHIVE" + archiveManifestObjectRecovery = "GC_RECOVERY" + archiveManifestVerified = "VERIFIED" +) + +// upsertArchiveManifest 以 COS key 为不可变对象身份。重验历史对象时只刷新校验结果， +// 不创建重复清单；GC 删除前可据此确认看到的是同一个对象版本和摘要。 +func (s *Service) upsertArchiveManifest(ctx context.Context, row model.TaskCenterArchiveManifest) error { + now := time.Now() + if row.CreateTime.IsZero() { + row.CreateTime = now + } + row.UpdateTime = now + return s.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "cos_key"}}, + DoUpdates: clause.Assignments(map[string]any{ + "object_type": row.ObjectType, + "run_id": row.RunID, + "parent_cos_key": row.ParentCOSKey, + "sha256": row.SHA256, + "etag": row.ETag, + "compressed_size": row.CompressedSize, + "row_count": row.RowCount, + "min_outbox_id": row.MinOutboxID, + "max_outbox_id": row.MaxOutboxID, + "min_occurred_at": row.MinOccurredAt, + "max_occurred_at": row.MaxOccurredAt, + "verification_status": row.VerificationStatus, + "verified_at": row.VerifiedAt, + "update_time": row.UpdateTime, + }), + }).Create(&row).Error +} + +func archiveGroupBounds(rows []model.TaskCenterEventArchiveOutbox) (*int64, *int64, *time.Time, *time.Time) { + if len(rows) == 0 { + return nil, nil, nil, nil + } + minID, maxID := rows[0].ID, rows[0].ID + var minOccurredAt, maxOccurredAt *time.Time + for _, row := range rows { + if row.ID < minID { + minID = row.ID + } + if row.ID > maxID { + maxID = row.ID + } + if row.OccurredAt.IsZero() { + continue + } + value := row.OccurredAt + if minOccurredAt == nil || value.Before(*minOccurredAt) { + copyValue := value + minOccurredAt = ©Value + } + if maxOccurredAt == nil || value.After(*maxOccurredAt) { + copyValue := value + maxOccurredAt = ©Value + } + } + return &minID, &maxID, minOccurredAt, maxOccurredAt +} diff --git a/internal/service/taskcenter/archive_object.go b/internal/service/taskcenter/archive_object.go new file mode 100644 index 0000000..f26f05d --- /dev/null +++ b/internal/service/taskcenter/archive_object.go @@ -0,0 +1,180 @@ +package taskcenter + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/md5" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "fmt" + "io" + "net/http" + "strconv" + "strings" + + "github.com/tencentyun/cos-go-sdk-v5" +) + +const ( + maxArchiveCOSObjectBytes = 128 << 20 + maxArchiveCOSUncompressedBytes = 256 << 20 +) + +type verifiedArchiveObject struct { + COSKey string + SHA256 string + ETag string + CompressedSize int64 + Header http.Header +} + +// putAndHeadArchiveObject 将本地摘要同时写入 COS metadata，并在 PUT 后重新 HEAD。 +// 只有服务端返回的长度和摘要都与本地一致才继续写 manifest/outbox，防止网关成功响应、 +// 错误对象 key 或被并发覆盖造成“数据库显示已归档但对象不可恢复”。 +func putAndHeadArchiveObject( + ctx context.Context, + client *cos.Client, + cosKey string, + body []byte, + metadata map[string]string, +) (verifiedArchiveObject, error) { + sha := archiveSHA256(body) + headers := make(http.Header) + headers.Set("x-cos-meta-sha256", sha) + for key, value := range metadata { + headers.Set("x-cos-meta-"+strings.ToLower(strings.TrimSpace(key)), strings.TrimSpace(value)) + } + response, err := client.Object.Put(ctx, cosKey, bytes.NewReader(body), &cos.ObjectPutOptions{ + ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{ + ContentType: "application/gzip", + ContentMD5: archiveContentMD5(body), + ContentLength: int64(len(body)), + XCosMetaXXX: &headers, + }, + }) + if response != nil && response.Body != nil { + _ = response.Body.Close() + } + if err != nil { + return verifiedArchiveObject{}, err + } + verified, err := headAndVerifyArchiveObject(ctx, client, cosKey, sha, int64(len(body)), true) + if err != nil { + return verifiedArchiveObject{}, err + } + for key, expected := range metadata { + headerName := "x-cos-meta-" + strings.ToLower(strings.TrimSpace(key)) + if actual := strings.TrimSpace(verified.Header.Get(headerName)); actual != strings.TrimSpace(expected) { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s metadata %s does not match", cosKey, headerName) + } + } + return verified, nil +} + +func headAndVerifyArchiveObject( + ctx context.Context, + client *cos.Client, + cosKey, expectedSHA string, + expectedSize int64, + requireSHAMetadata bool, +) (verifiedArchiveObject, error) { + response, err := client.Object.Head(ctx, cosKey, nil) + if response != nil && response.Body != nil { + defer response.Body.Close() + } + if err != nil { + return verifiedArchiveObject{}, err + } + if response == nil || response.Response == nil { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s returned no response", cosKey) + } + size, err := strconv.ParseInt(strings.TrimSpace(response.Header.Get("Content-Length")), 10, 64) + if err != nil || size < 0 { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s has invalid content length", cosKey) + } + if expectedSize >= 0 && size != expectedSize { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s size %d does not match %d", cosKey, size, expectedSize) + } + remoteSHA := strings.TrimSpace(response.Header.Get("x-cos-meta-sha256")) + if requireSHAMetadata && remoteSHA == "" { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s is missing required sha256 metadata", cosKey) + } + if remoteSHA != "" && expectedSHA != "" && !strings.EqualFold(remoteSHA, expectedSHA) { + return verifiedArchiveObject{}, fmt.Errorf("COS HEAD %s sha256 metadata mismatch", cosKey) + } + return verifiedArchiveObject{ + COSKey: cosKey, + SHA256: expectedSHA, + ETag: strings.Trim(strings.TrimSpace(response.Header.Get("ETag")), `"`), + CompressedSize: size, + Header: response.Header.Clone(), + }, nil +} + +// downloadArchiveObject 强制 identity 传输并限制压缩对象大小。清理路径必须拿到对象原始 +// gzip 字节才能同时验证 SHA256 和 JSONL，不能只相信 HEAD/ETag。 +func downloadArchiveObject(ctx context.Context, client *cos.Client, cosKey string) ([]byte, http.Header, error) { + headers := make(http.Header) + headers.Set("Accept-Encoding", "identity") + response, err := client.Object.Get(ctx, cosKey, &cos.ObjectGetOptions{XOptionHeader: &headers}) + if err != nil { + return nil, nil, err + } + if response == nil || response.Response == nil || response.Body == nil { + return nil, nil, fmt.Errorf("COS GET %s returned no body", cosKey) + } + defer response.Body.Close() + body, err := io.ReadAll(io.LimitReader(response.Body, maxArchiveCOSObjectBytes+1)) + if err != nil { + return nil, response.Header, err + } + if len(body) > maxArchiveCOSObjectBytes { + return nil, response.Header, fmt.Errorf("COS object %s exceeds %d bytes", cosKey, maxArchiveCOSObjectBytes) + } + return body, response.Header, nil +} + +// archiveObjectLines 解压并返回规范化 JSONL 行。历史对象与新对象都由同一个编码器写入； +// 空行被忽略，但单行上限和总解压大小都有硬门槛，避免损坏对象造成内存放大。 +func archiveObjectLines(body []byte) ([]string, error) { + if len(body) < 2 || body[0] != 0x1f || body[1] != 0x8b { + return nil, fmt.Errorf("archive object is not gzip encoded") + } + reader, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + return nil, err + } + defer reader.Close() + uncompressed, err := io.ReadAll(io.LimitReader(reader, maxArchiveCOSUncompressedBytes+1)) + if err != nil { + return nil, err + } + if len(uncompressed) > maxArchiveCOSUncompressedBytes { + return nil, fmt.Errorf("archive object expands beyond %d bytes", maxArchiveCOSUncompressedBytes) + } + parts := bytes.Split(uncompressed, []byte{'\n'}) + lines := make([]string, 0, len(parts)) + for _, part := range parts { + line := strings.TrimSpace(string(part)) + if line != "" { + lines = append(lines, line) + } + } + return lines, nil +} + +func archiveSHA256(value []byte) string { + sum := sha256.Sum256(value) + return hex.EncodeToString(sum[:]) +} + +func archiveContentMD5(value []byte) string { + sum := md5.Sum(value) + return base64.StdEncoding.EncodeToString(sum[:]) +} + +func archiveLineSHA256(value string) string { + return archiveSHA256([]byte(strings.TrimSpace(value))) +} diff --git a/migrations/060_task_center_archive_gc.sql b/migrations/060_task_center_archive_gc.sql new file mode 100644 index 0000000..b17e393 --- /dev/null +++ b/migrations/060_task_center_archive_gc.sql @@ -0,0 +1,65 @@ +-- Task center archive GC metadata only. These tables are deliberately small and do not alter the +-- 35M-row outbox; the heavy Game King secondary index remains the independently scheduled DDL in 058. +CREATE TABLE IF NOT EXISTS `task_center_archive_manifest` ( + `id` bigint NOT NULL AUTO_INCREMENT COMMENT 'manifest主键', + `object_type` varchar(32) NOT NULL COMMENT 'ARCHIVE或GC_RECOVERY', + `run_id` varchar(64) NOT NULL DEFAULT '' COMMENT 'GC运行ID，普通归档为空', + `cos_key` varchar(512) NOT NULL COMMENT 'COS对象Key', + `parent_cos_key` varchar(512) NOT NULL DEFAULT '' COMMENT '恢复包对应的原始归档Key', + `sha256` char(64) NOT NULL COMMENT '压缩对象SHA256', + `etag` varchar(128) NOT NULL DEFAULT '' COMMENT 'COS ETag', + `compressed_size` bigint NOT NULL COMMENT '压缩对象字节数', + `row_count` int NOT NULL COMMENT '对象JSONL行数', + `min_outbox_id` bigint DEFAULT NULL COMMENT '对象最小outbox ID；旧对象无法完整推导时为空', + `max_outbox_id` bigint DEFAULT NULL COMMENT '对象最大outbox ID；旧对象无法完整推导时为空', + `min_occurred_at` datetime(3) DEFAULT NULL COMMENT '对象最早业务时间', + `max_occurred_at` datetime(3) DEFAULT NULL COMMENT '对象最晚业务时间', + `verification_status` varchar(32) NOT NULL COMMENT 'VERIFIED或FAILED', + `verified_at` datetime(3) DEFAULT NULL COMMENT '最后一次完整回读校验时间', + `create_time` datetime(3) NOT NULL COMMENT '创建时间', + `update_time` datetime(3) NOT NULL COMMENT '更新时间', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_task_center_archive_manifest_cos_key` (`cos_key`), + KEY `idx_task_center_archive_manifest_run` (`run_id`, `object_type`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='任务中心COS归档与GC恢复包校验清单'; + +-- id 固定为 1：所有 API/consumer 实例锁定同一行取得数据库租约，避免多节点并发清理。 +CREATE TABLE IF NOT EXISTS `task_center_archive_gc_run` ( + `id` tinyint unsigned NOT NULL COMMENT '固定为1的全局控制行', + `run_id` varchar(64) NOT NULL DEFAULT '' COMMENT '本次固定cutoff运行ID', + `status` varchar(32) NOT NULL DEFAULT 'IDLE' COMMENT 'IDLE/DRY_RUN/RUNNING/PAUSED/COMPLETED/FAILED', + `cutoff_time` datetime(3) DEFAULT NULL COMMENT '预检时冻结的清理截止时间', + `batch_size` int NOT NULL DEFAULT '200' COMMENT '单tick删除上限，代码限制200到500', + `scanned_rows` bigint NOT NULL DEFAULT '0', + `eligible_rows` bigint NOT NULL DEFAULT '0', + `protected_rows` bigint NOT NULL DEFAULT '0', + `deleted_rows` bigint NOT NULL DEFAULT '0', + `recovery_object_count` bigint NOT NULL DEFAULT '0', + `cursor_create_time` datetime(3) DEFAULT NULL COMMENT '最后成功删除行的create_time，仅展示不用于跳过数据', + `cursor_id` bigint NOT NULL DEFAULT '0' COMMENT '最后成功删除行ID，仅展示不用于跳过数据', + `lease_owner` varchar(160) NOT NULL DEFAULT '' COMMENT '当前tick实例', + `lease_until` datetime(3) DEFAULT NULL COMMENT '短租约到期时间', + `last_error` varchar(1000) NOT NULL DEFAULT '', + `create_time` datetime(3) NOT NULL, + `update_time` datetime(3) NOT NULL, + PRIMARY KEY (`id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='任务中心归档GC全局控制与租约'; + +INSERT INTO `task_center_archive_gc_run` + (`id`, `status`, `batch_size`, `create_time`, `update_time`) +VALUES + (1, 'IDLE', 200, NOW(3), NOW(3)) +ON DUPLICATE KEY UPDATE `id` = VALUES(`id`); + +-- 补数水位让服务重启后仍从已确认游标继续；GC即使完成一批也不能依赖调用方内存判断历史是否补齐。 +CREATE TABLE IF NOT EXISTS `yumi_game_king_backfill_state` ( + `activity_id` bigint NOT NULL COMMENT '活动ID', + `last_outbox_id` bigint NOT NULL DEFAULT '0' COMMENT '最后确认的outbox ID', + `last_occurred_at` datetime(3) DEFAULT NULL COMMENT '与ID组成稳定复合游标', + `scanned_rows` bigint NOT NULL DEFAULT '0', + `accepted_rows` bigint NOT NULL DEFAULT '0', + `completed` tinyint(1) NOT NULL DEFAULT '0' COMMENT '最近一次扫描是否到达当前末尾', + `create_time` datetime(3) NOT NULL, + `update_time` datetime(3) NOT NULL, + PRIMARY KEY (`activity_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Yumi游戏王持久化补数水位';