fix(team_process): Clean up stuck processing owners on startup and batch completion

- Add CleanupStuckProcessingOwners database method to reset owners stuck in processing status back to valid state
- Call cleanup routine on application startup to recover from previous abnormal exits
- Call cleanup routine after batch processing completes to ensure no orphaned processing states remain
- Add markOwnerResult calls in processSingleTeam when team invitations are full to properly track owner status
- Prevents owners from being permanently stuck in processing state due to unexpected shutdowns or errors
This commit is contained in:
2026-02-07 02:20:50 +08:00
parent 60f57ab6a6
commit 40a9da0809
3 changed files with 23 additions and 0 deletions

View File

@@ -85,6 +85,11 @@ func main() {
fmt.Printf("%s[清理]%s 已清理 %d 个遗留的运行中批次记录\n", colorYellow, colorReset, affected) fmt.Printf("%s[清理]%s 已清理 %d 个遗留的运行中批次记录\n", colorYellow, colorReset, affected)
} }
// 启动时清理遗留的 processing 状态母号(上次异常退出可能残留)
if affected, err := database.Instance.CleanupStuckProcessingOwners(); err == nil && affected > 0 {
fmt.Printf("%s[清理]%s 已重置 %d 个残留的处理中母号为有效状态\n", colorYellow, colorReset, affected)
}
// 启动自动补号检查器(需在前端开启开关才会实际补号) // 启动自动补号检查器(需在前端开启开关才会实际补号)
api.StartAutoAddService() api.StartAutoAddService()

View File

@@ -367,6 +367,13 @@ func runTeamProcess(req TeamProcessRequest) {
defer func() { defer func() {
teamProcessState.Running = false teamProcessState.Running = false
// 清理残留的 processing 状态母号,重置为 valid
if database.Instance != nil {
if affected, err := database.Instance.CleanupStuckProcessingOwners(); err == nil && affected > 0 {
logger.Warning(fmt.Sprintf("批次结束,重置了 %d 个残留的处理中母号为有效状态", affected), "", "team")
}
}
// 无论任务是正常完成还是异常中断,都更新批次记录状态 // 无论任务是正常完成还是异常中断,都更新批次记录状态
if database.Instance != nil && batchID > 0 { if database.Instance != nil && batchID > 0 {
errorsStr := "" errorsStr := ""
@@ -993,6 +1000,7 @@ func processSingleTeam(idx int, req TeamProcessRequest) (result TeamProcessResul
result.AddedToS2A = int(atomic.LoadInt32(&s2aSuccessCount)) result.AddedToS2A = int(atomic.LoadInt32(&s2aSuccessCount))
result.Errors = append(result.Errors, "Team 邀请已满") result.Errors = append(result.Errors, "Team 邀请已满")
result.DurationMs = time.Since(startTime).Milliseconds() result.DurationMs = time.Since(startTime).Milliseconds()
markOwnerResult(result.AddedToS2A > 0)
return result return result
} }
@@ -1027,6 +1035,7 @@ func processSingleTeam(idx int, req TeamProcessRequest) (result TeamProcessResul
result.AddedToS2A = int(atomic.LoadInt32(&s2aSuccessCount)) result.AddedToS2A = int(atomic.LoadInt32(&s2aSuccessCount))
result.Errors = append(result.Errors, "Team 邀请已满") result.Errors = append(result.Errors, "Team 邀请已满")
result.DurationMs = time.Since(startTime).Milliseconds() result.DurationMs = time.Since(startTime).Milliseconds()
markOwnerResult(result.AddedToS2A > 0)
return result return result
} }

View File

@@ -627,6 +627,15 @@ func (d *DB) CleanupStuckBatchRuns() (int64, error) {
return result.RowsAffected() return result.RowsAffected()
} }
// CleanupStuckProcessingOwners 清理卡住的 processing 状态母号,重置为 valid
func (d *DB) CleanupStuckProcessingOwners() (int64, error) {
result, err := d.db.Exec(`UPDATE team_owners SET status = 'valid' WHERE status = 'processing'`)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
// GetBatchRunStats 获取批次统计 // GetBatchRunStats 获取批次统计
func (d *DB) GetBatchRunStats() map[string]interface{} { func (d *DB) GetBatchRunStats() map[string]interface{} {
stats := make(map[string]interface{}) stats := make(map[string]interface{})