366 lines
10 KiB
Go
366 lines
10 KiB
Go
// Copyright 2026 tamakyi. All rights reserved.
|
|
// Use of this source code is governed by a MIT-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package cmd
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/service/s3"
|
|
"github.com/sirupsen/logrus"
|
|
"github.com/urfave/cli/v2"
|
|
|
|
"github.com/tamakyi/TamaBox/internal/conf"
|
|
"github.com/tamakyi/TamaBox/internal/db"
|
|
"github.com/tamakyi/TamaBox/internal/storage"
|
|
)
|
|
|
|
const (
|
|
r2OutboxDefaultLimit = 100
|
|
r2OutboxWorkerInterval = time.Minute
|
|
r2OutboxInitialBackoff = time.Minute
|
|
r2OutboxMaxBackoff = time.Hour
|
|
r2OutboxMaxLastErrorSize = 1024
|
|
)
|
|
|
|
var errR2OutboxNotConfigured = errors.New("R2 configuration is incomplete")
|
|
|
|
// R2Outbox 有界处理一批持久化的 R2 对象删除任务。
|
|
var R2Outbox = &cli.Command{
|
|
Name: "r2-outbox",
|
|
Usage: "Drain a bounded batch of pending R2 object deletion tasks",
|
|
Action: runR2Outbox,
|
|
Flags: []cli.Flag{
|
|
&cli.IntFlag{
|
|
Name: "limit",
|
|
Value: r2OutboxDefaultLimit,
|
|
Usage: "Maximum number of due tasks to process",
|
|
},
|
|
},
|
|
}
|
|
|
|
type r2OutboxTaskStore interface {
|
|
ListDue(ctx context.Context, limit int) ([]*db.R2DeleteTask, error)
|
|
Delete(id uint) error
|
|
MarkFailed(id uint, nextRetry time.Time, lastError string) error
|
|
HasReferences(ctx context.Context, bucket, key string) (bool, error)
|
|
}
|
|
|
|
type r2ObjectDeleter interface {
|
|
DeleteObject(ctx context.Context, bucket, key string) error
|
|
}
|
|
|
|
type r2S3ObjectDeleter struct {
|
|
client *s3.Client
|
|
}
|
|
|
|
func (d *r2S3ObjectDeleter) DeleteObject(ctx context.Context, bucket, key string) error {
|
|
return storage.DeleteObject(ctx, d.client, bucket, key)
|
|
}
|
|
|
|
type r2OutboxResult struct {
|
|
Listed int
|
|
Removed int
|
|
ObjectsDeleted int
|
|
ReferenceOnly int
|
|
Failed int
|
|
Skipped int
|
|
Remaining int
|
|
}
|
|
|
|
// processR2Outbox 串行处理一批到期任务。单个任务失败不会中断同批后续任务；
|
|
// context 取消则立即停止，且不会把关闭过程中的取消误记为一次任务失败。
|
|
func processR2Outbox(ctx context.Context, store r2OutboxTaskStore, deleter r2ObjectDeleter, limit int) (result r2OutboxResult, err error) {
|
|
defer func() {
|
|
result.Remaining = result.Listed - result.Removed
|
|
}()
|
|
|
|
tasks, err := store.ListDue(ctx, limit)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return result, ctx.Err()
|
|
}
|
|
return result, fmt.Errorf("list R2 outbox tasks: %s", sanitizeR2OutboxError(err))
|
|
}
|
|
result.Listed = len(tasks)
|
|
|
|
for _, task := range tasks {
|
|
if err := ctx.Err(); err != nil {
|
|
return result, err
|
|
}
|
|
if task == nil {
|
|
result.Failed++
|
|
logrus.Warn("R2 outbox 返回了空任务，已跳过")
|
|
continue
|
|
}
|
|
|
|
logger := logrus.WithField("task_id", task.ID)
|
|
if task.Bucket != conf.Upload.ImageBucket {
|
|
result.Skipped++
|
|
logger.Warn("R2 outbox 任务 bucket 与当前配置不匹配，任务已保留")
|
|
continue
|
|
}
|
|
|
|
referenced, refErr := store.HasReferences(ctx, task.Bucket, task.Key)
|
|
if ctx.Err() != nil {
|
|
return result, ctx.Err()
|
|
}
|
|
if refErr != nil {
|
|
result.Failed++
|
|
markR2OutboxTaskFailed(store, task, refErr)
|
|
continue
|
|
}
|
|
if referenced {
|
|
// 还有引用时只清理过期 outbox 记录，绝不删除对象。最后一个引用被
|
|
// 删除时，对应数据库事务会重新入队一条任务。
|
|
if err := store.Delete(task.ID); err != nil {
|
|
result.Failed++
|
|
logger.WithField("error", sanitizeR2OutboxError(err)).Warn("R2 outbox 引用任务清理失败，任务已保留")
|
|
continue
|
|
}
|
|
result.Removed++
|
|
result.ReferenceOnly++
|
|
continue
|
|
}
|
|
|
|
deleteErr := deleter.DeleteObject(ctx, task.Bucket, task.Key)
|
|
if deleteErr != nil && !isS3ObjectNotFound(deleteErr) {
|
|
if ctx.Err() != nil {
|
|
return result, ctx.Err()
|
|
}
|
|
result.Failed++
|
|
markR2OutboxTaskFailed(store, task, deleteErr)
|
|
continue
|
|
}
|
|
|
|
// storage.DeleteObject 已将 404 归一为成功；这里再次识别 404，保证其他
|
|
// deleter 实现也遵循相同的幂等语义。只有任务记录也删除后才算完成。
|
|
if err := store.Delete(task.ID); err != nil {
|
|
result.Failed++
|
|
logger.WithField("error", sanitizeR2OutboxError(err)).Warn("R2 对象已删除，但 outbox 任务清理失败；后续将幂等重试")
|
|
continue
|
|
}
|
|
result.Removed++
|
|
result.ObjectsDeleted++
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func markR2OutboxTaskFailed(store r2OutboxTaskStore, task *db.R2DeleteTask, cause error) {
|
|
nextRetry := time.Now().Add(r2OutboxBackoff(task.Attempts))
|
|
lastError := sanitizeR2OutboxError(cause)
|
|
if err := store.MarkFailed(task.ID, nextRetry, lastError); err != nil {
|
|
logrus.WithFields(logrus.Fields{
|
|
"task_id": task.ID,
|
|
"processing_error": lastError,
|
|
"persist_error": sanitizeR2OutboxError(err),
|
|
}).Warn("R2 outbox 任务失败且重试状态写入失败，任务已保留")
|
|
return
|
|
}
|
|
logrus.WithFields(logrus.Fields{
|
|
"task_id": task.ID,
|
|
"error": lastError,
|
|
"next_retry": nextRetry,
|
|
}).Warn("R2 outbox 任务处理失败，已安排重试")
|
|
}
|
|
|
|
func r2OutboxBackoff(attempts int) time.Duration {
|
|
if attempts < 0 {
|
|
attempts = 0
|
|
}
|
|
// 1、2、4、8、16、32 分钟后，下一次本应到 64 分钟，直接封顶一小时。
|
|
if attempts >= 6 {
|
|
return r2OutboxMaxBackoff
|
|
}
|
|
return r2OutboxInitialBackoff << attempts
|
|
}
|
|
|
|
// sanitizeR2OutboxError 在持久化或记录错误前移除已知配置与凭据，并限制为
|
|
// 1024 字节。截断保持 UTF-8 边界，避免向 text 列写入半个多字节字符。
|
|
func sanitizeR2OutboxError(err error) string {
|
|
if err == nil {
|
|
return ""
|
|
}
|
|
message := err.Error()
|
|
sensitiveValues := []string{
|
|
conf.Database.DSN,
|
|
conf.Database.Password,
|
|
conf.Upload.ImageEndpoint,
|
|
conf.Upload.ImageAccessID,
|
|
conf.Upload.ImageAccessSecret,
|
|
conf.Upload.ImageBucket,
|
|
conf.Upload.ImageBucketCDNHost,
|
|
}
|
|
for _, value := range sensitiveValues {
|
|
if value != "" {
|
|
message = strings.ReplaceAll(message, value, "[redacted]")
|
|
}
|
|
}
|
|
if len(message) <= r2OutboxMaxLastErrorSize {
|
|
return message
|
|
}
|
|
end := r2OutboxMaxLastErrorSize
|
|
for end > 0 && !utf8.ValidString(message[:end]) {
|
|
end--
|
|
}
|
|
return message[:end]
|
|
}
|
|
|
|
func r2OutboxConfigured() bool {
|
|
return strings.TrimSpace(conf.Upload.ImageEndpoint) != "" &&
|
|
strings.TrimSpace(conf.Upload.ImageAccessID) != "" &&
|
|
strings.TrimSpace(conf.Upload.ImageAccessSecret) != "" &&
|
|
strings.TrimSpace(conf.Upload.ImageBucket) != ""
|
|
}
|
|
|
|
func newR2OutboxObjectDeleter(ctx context.Context) (r2ObjectDeleter, error) {
|
|
if !r2OutboxConfigured() {
|
|
return nil, errR2OutboxNotConfigured
|
|
}
|
|
client, err := storage.NewS3Client(
|
|
ctx,
|
|
conf.Upload.ImageEndpoint,
|
|
conf.Upload.ImageBucket,
|
|
conf.Upload.ImageAccessID,
|
|
conf.Upload.ImageAccessSecret,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("initialize R2 client: %s", sanitizeR2OutboxError(err))
|
|
}
|
|
return &r2S3ObjectDeleter{client: client}, nil
|
|
}
|
|
|
|
// startR2OutboxWorker 只能在 db.Init 成功后调用。goroutine 由调用方提供的
|
|
// WaitGroup 跟踪，Web 退出时先取消 ctx，再等待正在执行的一轮收尾。
|
|
func startR2OutboxWorker(ctx context.Context, wg *sync.WaitGroup) {
|
|
if !r2OutboxConfigured() {
|
|
return
|
|
}
|
|
deleter, err := newR2OutboxObjectDeleter(ctx)
|
|
if err != nil {
|
|
logrus.WithField("error", sanitizeR2OutboxError(err)).Warn("R2 outbox worker 启动失败，Web 服务继续运行")
|
|
return
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
runR2OutboxWorker(ctx, db.R2DeleteTasks, deleter)
|
|
}()
|
|
}
|
|
|
|
func runR2OutboxWorker(ctx context.Context, store r2OutboxTaskStore, deleter r2ObjectDeleter) {
|
|
ticker := time.NewTicker(r2OutboxWorkerInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
// 启动先立即跑一轮；处理期间不启动并发轮次。若一轮超过一分钟，
|
|
// ticker 只会合并唤醒，不会形成多个 worker 同时消费。
|
|
_, err := processR2Outbox(ctx, store, deleter, r2OutboxDefaultLimit)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
logrus.WithField("error", sanitizeR2OutboxError(err)).Warn("R2 outbox 本轮处理失败，将在下一分钟重试")
|
|
}
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func drainR2Outbox(ctx context.Context, limit int) (r2OutboxResult, error) {
|
|
deleter, err := newR2OutboxObjectDeleter(ctx)
|
|
if err != nil {
|
|
return r2OutboxResult{}, err
|
|
}
|
|
return processR2Outbox(ctx, db.R2DeleteTasks, deleter, limit)
|
|
}
|
|
|
|
func runR2Outbox(ctx *cli.Context) error {
|
|
limit := ctx.Int("limit")
|
|
if limit <= 0 {
|
|
return errors.New("--limit must be greater than zero")
|
|
}
|
|
if err := conf.Init(); err != nil {
|
|
return fmt.Errorf("load configuration: %w", err)
|
|
}
|
|
if !r2OutboxConfigured() {
|
|
return errR2OutboxNotConfigured
|
|
}
|
|
|
|
dbType := conf.Database.Type
|
|
var dsn string
|
|
switch dbType {
|
|
case "mysql", "":
|
|
dsn = conf.MySQLDsn()
|
|
case "postgres":
|
|
dsn = conf.PostgresDsn()
|
|
default:
|
|
return fmt.Errorf("unknown database type: %q", dbType)
|
|
}
|
|
conf.Database.DSN = dsn
|
|
gdb, err := db.Init(dbType, dsn)
|
|
if err != nil {
|
|
return fmt.Errorf("connect to database: %w", err)
|
|
}
|
|
if sqlDB, dbErr := gdb.DB(); dbErr == nil {
|
|
defer func() {
|
|
if err := sqlDB.Close(); err != nil {
|
|
logrus.WithField("error", sanitizeR2OutboxError(err)).Warn("关闭数据库连接池失败")
|
|
}
|
|
}()
|
|
}
|
|
|
|
result, err := drainR2Outbox(ctx.Context, limit)
|
|
if err != nil {
|
|
return fmt.Errorf("drain R2 outbox: %s", sanitizeR2OutboxError(err))
|
|
}
|
|
logR2OutboxResult("R2 outbox 本轮处理完成", result)
|
|
return nil
|
|
}
|
|
|
|
func logR2OutboxResult(message string, result r2OutboxResult) {
|
|
entry := logrus.WithFields(logrus.Fields{
|
|
"listed": result.Listed,
|
|
"removed": result.Removed,
|
|
"object_deletes": result.ObjectsDeleted,
|
|
"reference_only": result.ReferenceOnly,
|
|
"failed": result.Failed,
|
|
"skipped": result.Skipped,
|
|
"remaining": result.Remaining,
|
|
})
|
|
if result.Remaining > 0 {
|
|
entry.Warn(message + "，仍有任务待重试或保留")
|
|
return
|
|
}
|
|
entry.Info(message)
|
|
}
|
|
|
|
// drainR2OutboxAfterBan 在封禁事务提交后执行。R2 失败只影响对象清理进度，
|
|
// 不向 runBan 返回错误，避免运维误以为需要重新执行不可逆的数据库封禁。
|
|
func drainR2OutboxAfterBan(ctx context.Context) {
|
|
result, err := drainR2Outbox(ctx, r2OutboxDefaultLimit)
|
|
if err != nil {
|
|
logrus.WithField("error", sanitizeR2OutboxError(err)).Warn(
|
|
"数据库封禁与清理已提交，但 R2 outbox 本轮未能处理；任务仍保留，无需重跑 ban，可稍后运行 r2-outbox",
|
|
)
|
|
return
|
|
}
|
|
if result.Remaining > 0 {
|
|
logR2OutboxResult("数据库封禁与清理已提交；R2 对象清理将在后续重试，无需重跑 ban", result)
|
|
return
|
|
}
|
|
logR2OutboxResult("数据库封禁与清理已提交；R2 outbox 本轮处理完成", result)
|
|
}
|