可靠性: 完善 R2 备份恢复与对象生命周期
This commit is contained in:
tamakyi
@@ -0,0 +1,365 @@
|
||||
// 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)
|
||||
}
|
||||
Reference in New Issue
Block a user