diff --git a/backend/internal/app/chunk_quota_security_test.go b/backend/internal/app/chunk_quota_security_test.go new file mode 100644 index 000000000..806b76192 --- /dev/null +++ b/backend/internal/app/chunk_quota_security_test.go @@ -0,0 +1,58 @@ +package app + +import ( + "bytes" + "testing" + "time" +) + +func TestChunkQuotaRejectsSingleFileLimit(t *testing.T) { + svc := newResourceTestService(t) + if _, err := svc.reserveChunkedUploadQuota("user-1", megabytes(defaultRuntimePolicy().Resource.ResourceUploadMB)); err == nil { + t.Fatal("chunked upload accepted a file at the exclusive single-file limit") + } +} + +func TestChunkQuotaReservedUploadConsumesOnce(t *testing.T) { + svc := newResourceTestService(t) + if err := svc.ReserveChunkUploadSession("user", "session", 7, time.Now().Add(time.Hour), 32); err != nil { + t.Fatal(err) + } + day := time.Now().UTC().Format("2006-01-02") + if used, _ := svc.repo.DailyUploadBytes("user", day); used != 7 { + t.Fatalf("quota before receiving bytes=%d", used) + } + resource, err := svc.UploadReservedResourceFile("user", "session", "test.txt", 7, "file", 0, 0, 0, bytes.NewReader([]byte("payload")), "logical-key") + if err != nil { + t.Fatal(err) + } + if resource.Size != 7 { + t.Fatalf("size=%d", resource.Size) + } + if used, _ := svc.repo.DailyUploadBytes("user", day); used != 7 { + t.Fatalf("quota after completion=%d", used) + } + if err := svc.ReserveChunkUploadSession("user", "retry", 7, time.Now().Add(time.Hour), 32); err != nil { + t.Fatal(err) + } + reused, err := svc.UploadReservedResourceFile("user", "retry", "test.txt", 7, "file", 0, 0, 0, bytes.NewReader([]byte("payload")), "logical-key") + if err != nil || reused.ID != resource.ID { + t.Fatalf("retry=%v err=%v", reused, err) + } + if used, _ := svc.repo.DailyUploadBytes("user", day); used != 7 { + t.Fatalf("quota after replay=%d", used) + } +} + +func TestChunkQuotaFailedCompletionReleases(t *testing.T) { + svc := newResourceTestService(t) + if err := svc.ReserveChunkUploadSession("user", "failed", 7, time.Now().Add(time.Hour), 32); err != nil { + t.Fatal(err) + } + if _, err := svc.UploadReservedResourceFile("user", "failed", "test.txt", 7, "file", 0, 0, 0, nil); err == nil { + t.Fatal("nil file accepted") + } + if used, _ := svc.repo.DailyUploadBytes("user", time.Now().UTC().Format("2006-01-02")); used != 0 { + t.Fatalf("failed upload retained quota=%d", used) + } +} diff --git a/backend/internal/app/resource_oss_fallback_test.go b/backend/internal/app/resource_oss_fallback_test.go index cc944b004..7da11c47b 100644 --- a/backend/internal/app/resource_oss_fallback_test.go +++ b/backend/internal/app/resource_oss_fallback_test.go @@ -21,6 +21,7 @@ func newResourceFallbackTestService(t *testing.T) (*Service, *gorm.DB) { } if err := db.AutoMigrate( &model.Resource{}, + &model.UploadReservation{}, &model.UserOSSSetting{}, &model.SystemSetting{}, &model.StorageLocation{}, diff --git a/backend/internal/app/resource_store.go b/backend/internal/app/resource_store.go index f958f307d..44c235f64 100644 --- a/backend/internal/app/resource_store.go +++ b/backend/internal/app/resource_store.go @@ -11,6 +11,7 @@ import ( "errors" "fmt" "io" + "log" "mime/multipart" "net/http" "os" @@ -79,9 +80,24 @@ func (s *Service) uploadResource(userID string, header *multipart.FileHeader, ki } // UploadResourceFile 接收已完整落盘的本地文件(分片上传合并后调用)。 -// 它与 UploadResource 共享资源幂等、媒体探测、配额和持久化语义,唯一差异是分片会话已在 handler 校验单文件上限, -// 因而此处不再重复该上限检查;uploadIdentity 用于跨请求重试时复用同一逻辑资源,避免重复对象。 +// 它与 UploadResource 共享资源幂等、媒体探测、配额和持久化语义。 func (s *Service) UploadResourceFile(userID string, fileName string, size int64, kind string, width int, height int, durationMs int64, file io.ReadSeeker, uploadIdentity ...string) (*model.Resource, error) { + return s.uploadResourceFile(userID, "", fileName, size, kind, width, height, durationMs, file, uploadIdentity...) +} + +func (s *Service) UploadReservedResourceFile(userID, reservationID, fileName string, size int64, kind string, width, height int, durationMs int64, file io.ReadSeeker, uploadIdentity ...string) (*model.Resource, error) { + if reservationID == "" { + return nil, BadAuthRequest("上传会话不存在") + } + defer func() { + if err := s.ReleaseChunkUploadSession(userID, reservationID); err != nil { + log.Printf("release upload session failed: %v", err) + } + }() + return s.uploadResourceFile(userID, reservationID, fileName, size, kind, width, height, durationMs, file, uploadIdentity...) +} + +func (s *Service) uploadResourceFile(userID, reservationID, fileName string, size int64, kind string, width int, height int, durationMs int64, file io.ReadSeeker, uploadIdentity ...string) (*model.Resource, error) { if file == nil || size <= 0 { return nil, BadAuthRequest("请选择要上传的文件") } @@ -99,8 +115,17 @@ func (s *Service) UploadResourceFile(userID string, fileName string, size int64, } kind = uploadedResourceKind(mimeType) if existing != nil { + if reservationID != "" { + if err := s.ReleaseChunkUploadSession(userID, reservationID); err != nil { + return nil, err + } + } return s.retryStoredResource(userID, existing, kind, mimeType, size, file) } + if reservationID != "" { + resource, _, err := s.storeResource(userID, kind, fileName, mimeType, size, width, height, durationMs, file, uploadKey, false, reservationID) + return resource, err + } day, err := s.reserveChunkedUploadQuota(userID, size) if err != nil { return nil, err @@ -253,11 +278,15 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran return &ResourceStream{Resource: resource, Body: stream.Body, StatusCode: stream.StatusCode, ContentLength: stream.ContentLength, ContentRange: stream.ContentRange, AcceptRanges: stream.AcceptRanges}, nil } -func (s *Service) storeResource(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool) (*model.Resource, bool, error) { - return s.storeResourceWithWriter(userID, kind, fileName, mimeType, size, width, height, durationMs, body, uploadKey, forceLocal, s.storeResourceObject) +func (s *Service) storeResource(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool, reservationIDs ...string) (*model.Resource, bool, error) { + return s.storeResourceWithWriter(userID, kind, fileName, mimeType, size, width, height, durationMs, body, uploadKey, forceLocal, s.storeResourceObject, reservationIDs...) } -func (s *Service) storeResourceWithWriter(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool, writeObject func(*model.Resource, string, io.Reader) (string, error)) (*model.Resource, bool, error) { +func (s *Service) storeResourceWithWriter(userID string, kind string, fileName string, mimeType string, size int64, width int, height int, durationMs int64, body io.Reader, uploadKey *string, forceLocal bool, writeObject func(*model.Resource, string, io.Reader) (string, error), reservationIDs ...string) (*model.Resource, bool, error) { + reservationID := "" + if len(reservationIDs) > 0 { + reservationID = reservationIDs[0] + } // Live2D is reserved for the administrator's validated local import path. if !(forceLocal && kind == "live2d") { kind = normalizeResourceKind(kind, mimeType) @@ -321,7 +350,7 @@ func (s *Service) storeResourceWithWriter(userID string, kind string, fileName s } resource.Status = model.ResourceStatusReady resource.ETag = etag - if err := s.repo.SaveResource(&resource); err != nil { + if err := s.saveResourceWithinStorageLimit(&resource, reservationID); err != nil { cleanupErr := s.deleteStoredResourceObject(userID, &resource) if cleanupErr == nil { if deleteErr := s.repo.DeleteResource(userID, resource.ID); deleteErr != nil { @@ -435,7 +464,7 @@ func (s *Service) retryStoredResource(userID string, resource *model.Resource, k } resource.Status = model.ResourceStatusReady resource.ETag = etag - if err := s.repo.SaveResource(resource); err != nil { + if err := s.saveResourceWithinStorageLimit(resource, ""); err != nil { s.releaseRetryUploadQuota(userID, day, size) resource.Status = model.ResourceStatusFailed resource.Error = "保存资源重试就绪状态失败" diff --git a/backend/internal/app/resource_test.go b/backend/internal/app/resource_test.go index 1fb64ad40..fd3cb81b3 100644 --- a/backend/internal/app/resource_test.go +++ b/backend/internal/app/resource_test.go @@ -934,7 +934,7 @@ func newResourceTestService(t *testing.T) *Service { if err != nil { t.Fatal(err) } - if err := db.AutoMigrate(&model.SystemSetting{}, &model.UserOSSSetting{}, &model.StorageLocation{}, &model.UserDailyUploadUsage{}, &model.Resource{}); err != nil { + if err := db.AutoMigrate(&model.SystemSetting{}, &model.UserOSSSetting{}, &model.StorageLocation{}, &model.UserDailyUploadUsage{}, &model.Resource{}, &model.UploadReservation{}); err != nil { t.Fatal(err) } return &Service{repo: repository.New(db), dataDir: t.TempDir()} diff --git a/backend/internal/app/task_media_storage.go b/backend/internal/app/task_media_storage.go index df253be79..82610e247 100644 --- a/backend/internal/app/task_media_storage.go +++ b/backend/internal/app/task_media_storage.go @@ -266,7 +266,7 @@ func (s *Service) storeTaskMediaFile(task *model.Task, index int, path, mimeType resource.ETag, err = s.storeTaskMediaObject(resource, "generated."+extensionFromMimeType(mimeType), file) if err == nil { resource.Status, resource.Error, resource.UpdatedAt = model.ResourceStatusReady, "", time.Now() - err = s.repo.SaveResource(resource) + err = s.saveResourceWithinStorageLimit(resource, "") } } if err != nil { diff --git a/backend/internal/app/upload_quota.go b/backend/internal/app/upload_quota.go index da82588e0..1f29bde40 100644 --- a/backend/internal/app/upload_quota.go +++ b/backend/internal/app/upload_quota.go @@ -6,6 +6,7 @@ import ( "log" "time" + "infinite-canvas/backend/internal/model" "infinite-canvas/backend/internal/repository" ) @@ -18,14 +19,49 @@ func (s *Service) reserveUserUploadQuota(userID string, size int64) (string, err return s.reserveUserStoredFileQuota(userID, size, megabytes(policy.Resource.ResourceUploadMB), megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), fmt.Sprintf("单个上传文件必须小于 %dMB", policy.Resource.ResourceUploadMB)) } -// reserveChunkedUploadQuota 用于分片上传完成时预留额度:单文件上限对分片会话不适用(每片已独立校验), -// 仅受“今日上传”与“账号存储总量”约束;singleFileLimit 传 size+1 使单文件上限永不命中。 +// 分片和普通上传使用相同的单文件上限。 func (s *Service) reserveChunkedUploadQuota(userID string, size int64) (string, error) { + return s.reserveUserUploadQuota(userID, size) +} + +func (s *Service) ReserveChunkUploadSession(userID, id string, size int64, expires time.Time, maxSessions int) error { policy, err := s.RuntimePolicy() if err != nil { - return "", err + return err } - return s.reserveUserStoredFileQuota(userID, size, size+1, megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), "") + if size <= 0 || size >= megabytes(policy.Resource.ResourceUploadMB) { + return BadAuthRequest(fmt.Sprintf("单个上传文件必须小于 %dMB", policy.Resource.ResourceUploadMB)) + } + if !expires.After(time.Now()) || maxSessions <= 0 { + return BadAuthRequest("上传会话参数无效") + } + err = s.repo.ReserveUploadSession(&model.UploadReservation{ID: id, UserID: userID, Size: size, Day: time.Now().UTC().Format("2006-01-02"), ExpiresAt: expires}, megabytes(policy.Resource.DailyUploadMB), gigabytes(policy.Resource.StoredFileGB), maxSessions) + return uploadReservationError(err) +} + +func (s *Service) ReleaseChunkUploadSession(userID, id string) error { + return s.repo.ReleaseUploadSession(userID, id) +} + +func uploadReservationError(err error) error { + if errors.Is(err, repository.ErrUploadSessionLimit) { + return NewAppError(429, err.Error()) + } + if errors.Is(err, repository.ErrUploadStorageLimit) || errors.Is(err, repository.ErrDailyUploadLimitExceeded) { + return QuotaExceeded(err.Error()) + } + if errors.Is(err, repository.ErrUploadReservationExpired) { + return BadAuthRequest(err.Error()) + } + return err +} + +func (s *Service) saveResourceWithinStorageLimit(resource *model.Resource, reservationID string) error { + policy, err := s.RuntimePolicy() + if err != nil { + return err + } + return uploadReservationError(s.repo.SaveResourceWithinStorageLimit(resource, gigabytes(policy.Resource.StoredFileGB), megabytes(policy.Resource.DailyUploadMB), reservationID)) } func (s *Service) reserveGeneratedResourceQuota(userID string, size int64) (string, error) { diff --git a/backend/internal/database/migrations.go b/backend/internal/database/migrations.go index 302c7e916..518a27b98 100644 --- a/backend/internal/database/migrations.go +++ b/backend/internal/database/migrations.go @@ -146,6 +146,9 @@ var schemaMigrations = []migration{ {version: 43, name: "topup_sale_strategies", checksum: "sha256:topup-sale-strategies-v43-20260929", apply: func(tx *gorm.DB) error { return tx.AutoMigrate(&model.TopupProduct{}, &model.PaymentOrder{}) }}, + {version: 45, name: "upload_reservations", checksum: "sha256:upload-reservations-v45-20261005", apply: func(tx *gorm.DB) error { + return tx.AutoMigrate(&model.UploadReservation{}) + }}, {version: 46, name: "skill_curation", checksum: "sha256:skill-curation-v46-20261005", apply: migrateSkillCuration}, {version: 47, name: "skill_curation_roots", checksum: "sha256:skill-curation-roots-v47-20261005", apply: migrateSkillCurationRoots}, } diff --git a/backend/internal/database/migrations_test.go b/backend/internal/database/migrations_test.go index b96fb150d..ff0d526a7 100644 --- a/backend/internal/database/migrations_test.go +++ b/backend/internal/database/migrations_test.go @@ -36,6 +36,9 @@ func TestMigrateSchemaRecordsAndValidatesVersion(t *testing.T) { if !status.Ready || status.Current != CurrentSchemaVersion { t.Fatalf("unexpected schema status: %#v", status) } + if !db.Migrator().HasTable(&model.UploadReservation{}) { + t.Fatal("upload reservation migration missing") + } if !db.Migrator().HasIndex(&schemaMigration{}, "idx_schema_migrations_applied_at") { t.Fatal("schema migration v2 did not create the applied_at index") } diff --git a/backend/internal/database/schema.go b/backend/internal/database/schema.go index 0f982a32b..d6f56d39e 100644 --- a/backend/internal/database/schema.go +++ b/backend/internal/database/schema.go @@ -17,6 +17,7 @@ import ( func Models() []any { return []any{ &model.CloudAgentExecution{}, + &model.UploadReservation{}, &model.CloudAgentPiSession{}, &model.CloudAgentEventRecord{}, &model.CloudAgentMessageRecord{}, diff --git a/backend/internal/handler/chunk_quota_security_test.go b/backend/internal/handler/chunk_quota_security_test.go new file mode 100644 index 000000000..90edba801 --- /dev/null +++ b/backend/internal/handler/chunk_quota_security_test.go @@ -0,0 +1,108 @@ +package handler + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "infinite-canvas/backend/internal/auth" + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/platform" + "infinite-canvas/backend/internal/repository" + "infinite-canvas/backend/internal/service" +) + +func TestChunkAdmissionReservesBeforeBody(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + sqlDB, _ := db.DB() + sqlDB.SetMaxOpenConns(1) + t.Cleanup(func() { _ = sqlDB.Close() }) + if err := db.AutoMigrate(&model.User{}, &model.AuthSession{}, &model.SystemSetting{}, &model.Resource{}, &model.UploadReservation{}, &model.UserDailyUploadUsage{}, &model.StorageLocation{}, &model.UserOSSSetting{}, &model.UserDailyActivity{}); err != nil { + t.Fatal(err) + } + if err := db.Create(&model.User{ID: "chunk-user", Username: "chunk-user", Status: model.UserStatusActive}).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&model.AuthSession{ID: "chunk-session", UserID: "chunk-user", TokenHash: auth.HashToken("test-token"), ExpiresAt: time.Now().Add(time.Hour)}).Error; err != nil { + t.Fatal(err) + } + policy := platform.DefaultRuntimePolicy() + policy.Resource.ResourceUploadMB = 1 + data, _ := json.Marshal(policy) + if err := db.Create(&model.SystemSetting{Key: "runtime_policy", ValueJSON: string(data)}).Error; err != nil { + t.Fatal(err) + } + repo := repository.New(db) + svc := service.New(repo, t.TempDir()) + previous := runtimeService + ConfigureRuntime(svc) + t.Cleanup(func() { runtimeService = previous }) + gin.SetMode(gin.TestMode) + router := gin.New() + RegisterChunkedUploadRoutes(router.Group("/api"), svc) + call := func(method, path, body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(method, path, strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.AddCookie(&http.Cookie{Name: service.SessionCookieName, Value: "chunk-session.test-token"}) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + return w + } + w := call("POST", "/api/resources/uploads", `{"fileName":"large.txt","size":1048576}`) + if w.Code != 400 { + t.Fatalf("oversize admission status=%d body=%s", w.Code, w.Body.String()) + } + w = call("POST", "/api/resources/uploads", `{"fileName":"test.txt","size":7}`) + if w.Code != 200 { + t.Fatalf("admission status=%d body=%s", w.Code, w.Body.String()) + } + var response struct { + Data struct { + UploadID string `json:"uploadId"` + } `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + id := response.Data.UploadID + t.Cleanup(func() { + if sess := takeChunkSession(id); sess != nil { + sess.mu.Lock() + dropChunkSession(id) + sess.mu.Unlock() + } + }) + if used, _ := repo.DailyUploadBytes("chunk-user", time.Now().UTC().Format("2006-01-02")); used != 7 { + t.Fatalf("before first chunk daily bytes=%d want 7", used) + } + var count int64 + if err := db.Model(&model.UploadReservation{}).Where("id = ?", id).Count(&count).Error; err != nil || count != 1 { + t.Fatalf("admission reservation count=%d err=%v", count, err) + } + w = call("PUT", fmt.Sprintf("/api/resources/uploads/%s/chunks/0", id), "payload") + if w.Code != 200 { + t.Fatalf("chunk status=%d body=%s", w.Code, w.Body.String()) + } + w = call("POST", fmt.Sprintf("/api/resources/uploads/%s/complete", id), "") + if w.Code != 200 || !bytes.Contains(w.Body.Bytes(), []byte(`"resource"`)) { + t.Fatalf("complete status=%d body=%s", w.Code, w.Body.String()) + } + if err := db.Model(&model.UploadReservation{}).Count(&count).Error; err != nil || count != 0 { + t.Fatalf("remaining reservations=%d err=%v", count, err) + } + w = call("POST", fmt.Sprintf("/api/resources/uploads/%s/complete", id), "") + if w.Code != 404 { + t.Fatalf("duplicate completion status=%d", w.Code) + } +} diff --git a/backend/internal/handler/resource_upload_session.go b/backend/internal/handler/resource_upload_session.go index 4e824c64f..5a117b916 100644 --- a/backend/internal/handler/resource_upload_session.go +++ b/backend/internal/handler/resource_upload_session.go @@ -5,6 +5,7 @@ import ( "encoding/hex" "fmt" "io" + "log" "net/http" "os" "path/filepath" @@ -18,8 +19,8 @@ import ( ) // 分片上传会话:把“导入本地媒体”拆成 开始→逐片→合并 三段,单片上限 8MB, -// 文件整体不再受 multipart 单请求大小限制(对齐 Concat 桌面端“任意大小直接入库”的体验)。 -// 会话状态只存在内存(重启即失效 → 前端整传重试),磁盘暂存在系统临时目录,随会话清理。 +// 整体文件仍遵守资源单文件上限;收片前在数据库预留日额度和存储容量。 +// 分片路由需要粘性路由,内存会话失效后重传;配额预留跨实例共享并按时间到期。 const ( chunkUploadChunkSize = 8 << 20 chunkUploadSlackBytes = 64 << 10 // MaxBytesReader 允许的超片余量 @@ -29,6 +30,9 @@ const ( ) type chunkedUploadSession struct { + mu sync.Mutex + closed bool + release func() error ID string UserID string FileName string @@ -87,9 +91,16 @@ func removeExpiredChunkSessions() { chunkUploadSessions.Lock() defer chunkUploadSessions.Unlock() for id, sess := range chunkUploadSessions.m { - if now.Sub(sess.CreatedAt) > chunkUploadTTL { + if now.Sub(sess.CreatedAt) > chunkUploadTTL && sess.mu.TryLock() { + sess.closed = true + if sess.release != nil { + if err := sess.release(); err != nil { + log.Printf("release expired upload session failed: %v", err) + } + } _ = os.RemoveAll(sess.Dir) delete(chunkUploadSessions.m, id) + sess.mu.Unlock() } } } @@ -105,6 +116,13 @@ func dropChunkSession(id string) { chunkUploadSessions.Lock() defer chunkUploadSessions.Unlock() if sess := chunkUploadSessions.m[id]; sess != nil { + // Caller holds sess.mu. Expiry uses TryLock to avoid lock inversion. + sess.closed = true + if sess.release != nil { + if err := sess.release(); err != nil { + log.Printf("release upload session failed: %v", err) + } + } _ = os.RemoveAll(sess.Dir) delete(chunkUploadSessions.m, id) } @@ -144,31 +162,24 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusBadRequest, fmt.Errorf("文件大小必须大于 0")) return } - // 超账号存储总量的文件无论如何都会失败,提前给出明确提示。 - if policy.Resource.StoredFileGB > 0 && req.Size > int64(policy.Resource.StoredFileGB)<<30 { - fail(c, http.StatusBadRequest, fmt.Errorf("文件超过账号存储总量上限 %dGB", policy.Resource.StoredFileGB)) - return - } - // 同一用户并发会话数兜底,防内存占用失控。 - active := 0 - chunkUploadSessions.Lock() - for _, sess := range chunkUploadSessions.m { - if sess.UserID == user.ID { - active++ - } - } - chunkUploadSessions.Unlock() - if active >= chunkUploadMaxPerUser { - fail(c, http.StatusTooManyRequests, fmt.Errorf("同时进行中的上传过多,请稍后重试")) + removeExpiredChunkSessions() + id, createdAt := newUploadSessionID(), time.Now() + if err := svc.ReserveChunkUploadSession(user.ID, id, req.Size, createdAt.Add(chunkUploadTTL), chunkUploadMaxPerUser); err != nil { + failService(c, err) return } + release := func() error { return svc.ReleaseChunkUploadSession(user.ID, id) } dir, err := os.MkdirTemp("", "canvas-chunk-upload-*") if err != nil { + if releaseErr := release(); releaseErr != nil { + log.Printf("release upload session failed: %v", releaseErr) + } failService(c, err) return } session := &chunkedUploadSession{ - ID: newUploadSessionID(), + ID: id, + release: release, UserID: user.ID, FileName: req.FileName, Kind: req.Kind, @@ -179,7 +190,7 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { IdempotencyKey: req.IdempotencyKey, ChunkCount: int((req.Size + chunkUploadChunkSize - 1) / chunkUploadChunkSize), Dir: dir, - CreatedAt: time.Now(), + CreatedAt: createdAt, } chunkUploadSessions.Lock() chunkUploadSessions.m[session.ID] = session @@ -199,6 +210,12 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { return } index, err := strconv.Atoi(c.Param("index")) + session.mu.Lock() + defer session.mu.Unlock() + if session.closed || time.Since(session.CreatedAt) >= chunkUploadTTL { + fail(c, http.StatusNotFound, fmt.Errorf("上传会话不存在或已过期,请重新导入")) + return + } if err != nil || index < 0 || index >= session.ChunkCount { fail(c, http.StatusBadRequest, fmt.Errorf("非法的分片序号")) return @@ -224,7 +241,7 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusBadRequest, fmt.Errorf("分片 %d 上传不完整,请重试", index)) return } - if readErr == nil || extra > 0 { + if readErr != io.EOF || extra > 0 { _ = os.Remove(session.chunkPath(index)) fail(c, http.StatusBadRequest, fmt.Errorf("分片 %d 超过大小限制", index)) return @@ -245,6 +262,12 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { fail(c, http.StatusNotFound, fmt.Errorf("上传会话不存在或已过期,请重新导入")) return } + session.mu.Lock() + defer session.mu.Unlock() + if session.closed || time.Since(session.CreatedAt) >= chunkUploadTTL { + fail(c, http.StatusNotFound, fmt.Errorf("上传会话不存在或已过期,请重新导入")) + return + } if !session.hasAllChunks() { fail(c, http.StatusBadRequest, fmt.Errorf("上传文件不完整,请重新导入")) return @@ -287,7 +310,7 @@ func RegisterChunkedUploadRoutes(r *gin.RouterGroup, svc *service.Service) { return } defer fh.Close() - resource, svcErr := svc.UploadResourceFile(user.ID, session.FileName, session.Size, session.Kind, session.Width, session.Height, session.DurationMs, fh, session.IdempotencyKey) + resource, svcErr := svc.UploadReservedResourceFile(user.ID, session.ID, session.FileName, session.Size, session.Kind, session.Width, session.Height, session.DurationMs, fh, session.IdempotencyKey) // 无论成败都结束会话:失败时前端会整传重试,不需要保留残片。 dropChunkSession(id) if svcErr != nil { diff --git a/backend/internal/model/upload_reservation.go b/backend/internal/model/upload_reservation.go new file mode 100644 index 000000000..5b1d8db20 --- /dev/null +++ b/backend/internal/model/upload_reservation.go @@ -0,0 +1,13 @@ +package model + +import "time" + +// UploadReservation survives process restarts; expiry, not process ownership, +// bounds its lifetime. Day records the UTC quota bucket charged at admission. +type UploadReservation struct { + ID string `gorm:"primaryKey;size:64"` + UserID string `gorm:"index;size:64;not null"` + Size int64 `gorm:"not null"` + Day string `gorm:"size:10;not null"` + ExpiresAt time.Time `gorm:"index;not null"` +} diff --git a/backend/internal/repository/upload_reservations.go b/backend/internal/repository/upload_reservations.go new file mode 100644 index 000000000..43f27efa2 --- /dev/null +++ b/backend/internal/repository/upload_reservations.go @@ -0,0 +1,141 @@ +package repository + +import ( + "errors" + "time" + + "gorm.io/gorm" + "infinite-canvas/backend/internal/model" +) + +var ErrUploadStorageLimit = errors.New("账号存储容量不足") +var ErrUploadSessionLimit = errors.New("同时进行中的上传过多,请稍后重试") +var ErrUploadReservationExpired = errors.New("上传会话不存在或已过期,请重新导入") + +func lockUploadStorage(tx *gorm.DB, userID string) error { + switch tx.Dialector.Name() { + case "postgres": + return tx.Exec("SELECT pg_advisory_xact_lock(hashtext(?)::bigint)", "upload-storage:"+userID).Error + case "sqlite": + return tx.Exec("UPDATE upload_reservations SET size = size WHERE 1 = 0").Error + default: + return gorm.ErrInvalidData + } +} + +func (r *Repository) ReserveUploadSession(reservation *model.UploadReservation, dailyLimit, storedLimit int64, sessionLimit int) error { + return r.db.Transaction(func(tx *gorm.DB) error { + if err := lockUploadStorage(tx, reservation.UserID); err != nil { + return err + } + repo := New(tx) + now := time.Now() + if !reservation.ExpiresAt.After(now) { + return ErrUploadReservationExpired + } + reservation.Day = now.UTC().Format("2006-01-02") + var expired []model.UploadReservation + if err := tx.Where("user_id = ? AND expires_at <= ?", reservation.UserID, now).Find(&expired).Error; err != nil { + return err + } + for _, old := range expired { + if err := repo.ReleaseDailyUpload(old.UserID, old.Day, old.Size); err != nil { + return err + } + if err := tx.Delete(&old).Error; err != nil { + return err + } + } + var active int64 + if err := tx.Model(&model.UploadReservation{}).Where("user_id = ?", reservation.UserID).Count(&active).Error; err != nil { + return err + } + if active >= int64(sessionLimit) { + return ErrUploadSessionLimit + } + stored, err := repo.UserStoredFileBytes(reservation.UserID) + if err != nil { + return err + } + var pending int64 + if err := tx.Model(&model.UploadReservation{}).Select("COALESCE(SUM(size),0)").Where("user_id = ?", reservation.UserID).Scan(&pending).Error; err != nil { + return err + } + if reservation.Size <= 0 || stored >= storedLimit || pending >= storedLimit-stored || reservation.Size >= storedLimit-stored-pending { + return ErrUploadStorageLimit + } + if err := repo.ReserveDailyUpload(reservation.UserID, reservation.Day, reservation.Size, dailyLimit); err != nil { + return err + } + return tx.Create(reservation).Error + }) +} + +func (r *Repository) ReleaseUploadSession(userID, id string) error { + return r.db.Transaction(func(tx *gorm.DB) error { + if err := lockUploadStorage(tx, userID); err != nil { + return err + } + var reservation model.UploadReservation + if err := tx.Where("id = ? AND user_id = ?", id, userID).First(&reservation).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + return err + } + if err := New(tx).ReleaseDailyUpload(userID, reservation.Day, reservation.Size); err != nil { + return err + } + return tx.Delete(&reservation).Error + }) +} + +// SaveResourceWithinStorageLimit publishes ready bytes and consumes the session +// reservation under the same database lock used by admission on every instance. +func (r *Repository) SaveResourceWithinStorageLimit(resource *model.Resource, limit, dailyLimit int64, reservationID string) error { + return r.db.Transaction(func(tx *gorm.DB) error { + if err := lockUploadStorage(tx, resource.UserID); err != nil { + return err + } + now := time.Now() + if reservationID != "" { + var reservation model.UploadReservation + if err := tx.Where("id = ? AND user_id = ? AND size = ? AND expires_at > ?", reservationID, resource.UserID, resource.Size, now).First(&reservation).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return ErrUploadReservationExpired + } + return err + } + day := now.UTC().Format("2006-01-02") + if reservation.Day != day { + repo := New(tx) + if err := repo.ReserveDailyUpload(resource.UserID, day, reservation.Size, dailyLimit); err != nil { + return err + } + if err := repo.ReleaseDailyUpload(resource.UserID, reservation.Day, reservation.Size); err != nil { + return err + } + } + } + // Save inside the transaction so physical-object deduplication uses the + // existing usage query; rollback leaves the resource non-ready on denial. + if err := tx.Save(resource).Error; err != nil { + return err + } + stored, err := New(tx).UserStoredFileBytes(resource.UserID) + if err != nil { + return err + } + var pending int64 + if err := tx.Model(&model.UploadReservation{}).Select("COALESCE(SUM(size),0)").Where("user_id = ? AND id <> ? AND expires_at > ?", resource.UserID, reservationID, now).Scan(&pending).Error; err != nil { + return err + } + if stored >= limit || pending >= limit-stored { + return ErrUploadStorageLimit + } + if reservationID != "" { + return tx.Delete(&model.UploadReservation{}, "id = ? AND user_id = ?", reservationID, resource.UserID).Error + } + return nil + }) +} diff --git a/backend/internal/repository/upload_reservations_test.go b/backend/internal/repository/upload_reservations_test.go new file mode 100644 index 000000000..507992b5b --- /dev/null +++ b/backend/internal/repository/upload_reservations_test.go @@ -0,0 +1,195 @@ +package repository + +import ( + "errors" + "fmt" + "path/filepath" + "sync" + "testing" + "time" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" + "infinite-canvas/backend/internal/model" +) + +func uploadReservationTestDB(t *testing.T) (*gorm.DB, *Repository, *Repository) { + t.Helper() + dsn := filepath.Join(t.TempDir(), "uploads.db") + "?_busy_timeout=5000&_journal_mode=WAL" + db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.UploadReservation{}, &model.UserDailyUploadUsage{}, &model.Resource{}); err != nil { + t.Fatal(err) + } + peer, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + for _, conn := range []*gorm.DB{db, peer} { + sqlDB, _ := conn.DB() + t.Cleanup(func() { _ = sqlDB.Close() }) + } + return db, New(db), New(peer) +} + +func TestUploadReservationConcurrentAdmission(t *testing.T) { + for _, tc := range []struct { + name string + size, daily, stored int64 + sessions, attempts, want int + denied error + }{ + {"sessions", 1, 1000, 1000, 32, 40, 32, ErrUploadSessionLimit}, + {"storage", 60, 1000, 100, 32, 2, 1, ErrUploadStorageLimit}, + {"daily", 60, 100, 1000, 32, 2, 1, ErrDailyUploadLimitExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + _, first, second := uploadReservationTestDB(t) + start := make(chan struct{}) + results := make(chan error, tc.attempts) + var wg sync.WaitGroup + for i := 0; i < tc.attempts; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + repo := []*Repository{first, second}[i%2] + results <- repo.ReserveUploadSession(&model.UploadReservation{ID: fmt.Sprint(i), UserID: "user", Size: tc.size, ExpiresAt: time.Now().Add(time.Hour)}, tc.daily, tc.stored, tc.sessions) + }() + } + close(start) + wg.Wait() + close(results) + accepted := 0 + for err := range results { + if err == nil { + accepted++ + } else if !errors.Is(err, tc.denied) { + t.Fatalf("unexpected denial: %v", err) + } + } + if accepted != tc.want { + t.Fatalf("accepted=%d want=%d", accepted, tc.want) + } + if used, err := first.DailyUploadBytes("user", time.Now().UTC().Format("2006-01-02")); err != nil || used != int64(tc.want)*tc.size { + t.Fatalf("reserved bytes=%d err=%v", used, err) + } + }) + } +} + +func TestUploadReservationAtomicCapacityCommit(t *testing.T) { + db, first, second := uploadReservationTestDB(t) + start := make(chan struct{}) + results := make(chan error, 2) + var wg sync.WaitGroup + for i, repo := range []*Repository{first, second} { + wg.Add(1) + go func() { + defer wg.Done() + <-start + results <- repo.SaveResourceWithinStorageLimit(&model.Resource{ID: fmt.Sprint(i), UserID: "user", ObjectKey: fmt.Sprint(i), Status: model.ResourceStatusReady, Size: 60}, 100, 1000, "") + }() + } + close(start) + wg.Wait() + close(results) + accepted := 0 + for err := range results { + if err == nil { + accepted++ + } else if !errors.Is(err, ErrUploadStorageLimit) { + t.Fatal(err) + } + } + if accepted != 1 { + t.Fatalf("committed=%d want=1", accepted) + } + var count int64 + if err := db.Model(&model.Resource{}).Count(&count).Error; err != nil || count != 1 { + t.Fatalf("resource rows=%d err=%v", count, err) + } +} + +func TestUploadReservationCommitAndExpiry(t *testing.T) { + db, r, peer := uploadReservationTestDB(t) + old := model.UploadReservation{ID: "old", UserID: "user", Size: 60, Day: "2026-10-04", ExpiresAt: time.Now().Add(-time.Hour)} + if err := db.Create(&old).Error; err != nil { + t.Fatal(err) + } + if err := r.ReserveDailyUpload("user", old.Day, 60, 1000); err != nil { + t.Fatal(err) + } + lease := model.UploadReservation{ID: "new", UserID: "user", Size: 60, Day: "2026-10-05", ExpiresAt: time.Now().Add(time.Hour)} + if err := peer.ReserveUploadSession(&lease, 1000, 100, 1); err != nil { + t.Fatal(err) + } + if used, _ := r.DailyUploadBytes("user", old.Day); used != 0 { + t.Fatalf("expired previous-day bytes=%d", used) + } + if err := r.ReleaseUploadSession("other", lease.ID); err != nil { + t.Fatal(err) + } + other := model.Resource{ID: "other", UserID: "user", Size: 40, Status: model.ResourceStatusReady, ObjectKey: "other"} + if err := r.SaveResourceWithinStorageLimit(&other, 100, 1000, ""); !errors.Is(err, ErrUploadStorageLimit) { + t.Fatalf("ordinary upload ignored reserved capacity: %v", err) + } + resource := model.Resource{ID: "resource", UserID: "user", Size: 59, Status: model.ResourceStatusReady, ObjectKey: "resource"} + if err := r.SaveResourceWithinStorageLimit(&resource, 100, 1000, lease.ID); !errors.Is(err, ErrUploadReservationExpired) { + t.Fatalf("mismatched lease: %v", err) + } + resource.Size = 60 + if err := r.SaveResourceWithinStorageLimit(&resource, 100, 1000, lease.ID); err != nil { + t.Fatal(err) + } + if err := peer.ReleaseUploadSession("user", lease.ID); err != nil { + t.Fatal(err) + } + if used, _ := r.DailyUploadBytes("user", lease.Day); used != 60 { + t.Fatalf("committed daily bytes=%d", used) + } + if err := r.SaveResourceWithinStorageLimit(&resource, 100, 1000, lease.ID); !errors.Is(err, ErrUploadReservationExpired) { + t.Fatalf("lease reused: %v", err) + } + resource.ID = "expired" + if err := r.SaveResourceWithinStorageLimit(&resource, 100, 1000, old.ID); !errors.Is(err, ErrUploadReservationExpired) { + t.Fatalf("expired lease accepted: %v", err) + } +} + +func TestUploadReservationCommitAcrossUTCDay(t *testing.T) { + db, r, _ := uploadReservationTestDB(t) + today := time.Now().UTC().Format("2006-01-02") + yesterday := time.Now().UTC().Add(-24 * time.Hour).Format("2006-01-02") + lease := model.UploadReservation{ID: "midnight", UserID: "user", Size: 15, Day: yesterday, ExpiresAt: time.Now().Add(time.Hour)} + if err := db.Create(&lease).Error; err != nil { + t.Fatal(err) + } + if err := r.ReserveDailyUpload("user", yesterday, 15, 100); err != nil { + t.Fatal(err) + } + if err := r.ReserveDailyUpload("user", today, 90, 100); err != nil { + t.Fatal(err) + } + resource := model.Resource{ID: "midnight", UserID: "user", Size: 15, ObjectKey: "midnight", Status: model.ResourceStatusReady} + if err := r.SaveResourceWithinStorageLimit(&resource, 1000, 100, lease.ID); !errors.Is(err, ErrDailyUploadLimitExceeded) { + t.Fatalf("new-day quota bypassed: %v", err) + } + if used, _ := r.DailyUploadBytes("user", yesterday); used != 15 { + t.Fatalf("failed commit refunded reservation: %d", used) + } + if err := r.ReleaseDailyUpload("user", today, 90); err != nil { + t.Fatal(err) + } + if err := r.SaveResourceWithinStorageLimit(&resource, 1000, 100, lease.ID); err != nil { + t.Fatal(err) + } + if used, _ := r.DailyUploadBytes("user", today); used != 15 { + t.Fatalf("new-day usage=%d", used) + } + if used, _ := r.DailyUploadBytes("user", yesterday); used != 0 { + t.Fatalf("old-day reservation=%d", used) + } +} diff --git a/docs/content/docs/backend/backend-database.mdx b/docs/content/docs/backend/backend-database.mdx index 9874dcaa6..4211cb0e6 100644 --- a/docs/content/docs/backend/backend-database.mdx +++ b/docs/content/docs/backend/backend-database.mdx @@ -5,6 +5,14 @@ description: 影策后端核心持久化表与状态边界 # 后端数据库 +## 分片上传预留 + +Schema 45 新增 `upload_reservations`:`id`、`user_id`、`size`、`day`、`expires_at`。创建会话时在数据库事务内检查并行会话数、现有资源容量、未到期预留和每日上传用量,更新既有 `user_daily_upload_usages` 后插入预留。SQLite 先取得写锁,PostgreSQL 使用账号级事务 advisory lock。 + +资源变为 ready 与删除对应预留在同一事务中提交;普通资源上传及生成结果的容量提交也使用同一锁,并计入分片预留。沿用 `UserStoredFileBytes` 的 ready 物理对象去重口径,不改变容量配置。跨 UTC 日期完成时重新预留当天日额度并释放原日额度,失败则整体回滚。 + +预留有效期 90 分钟。失败或会话删除时释放原日额度;再次创建该账号会话时清理已到期预留。进程重启不批量删除其他实例的预留。分片文件与路由会话仍在本机,需要粘性路由;崩溃后的残片需要部署侧临时目录清理策略,未增加跨实例文件恢复或启动时全局清扫。 + ## 更新公告配置 外观 JSON 合同版本 10 新增 `updates`:`enabled`、`currentVersion`、`releases`。每个版本保存 `id/version/title/date/blocks`,内容块保存 `id/type/text/resourceId/caption/layout`;公共媒体 URL 由服务端按当前引用生成,不保存客户端 URL。关闭开关保留公告和资源引用,只关闭公共展示。不增加数据库表或 schema 迁移。 diff --git a/docs/content/docs/backend/http-api.mdx b/docs/content/docs/backend/http-api.mdx index 92f232c7d..ea1397648 100644 --- a/docs/content/docs/backend/http-api.mdx +++ b/docs/content/docs/backend/http-api.mdx @@ -11,6 +11,12 @@ description: 信封、错误码与 OpenAPI 邮箱原值仍按原有小写、去首尾空格规则保存。登录、密码重置、验证码收件地址保持精确邮箱匹配,避免既有重复别名账号被合并或错误授权。已存在的重复账号不迁移、不删除、不重新发放注册奖励;它们会阻止相同规范信箱的新注册。写入前在事务内重新检查,以覆盖不同实例使用不同验证码的并发注册。无数据库结构迁移;唯一性由公开注册与绑定的事务入口保证,直接 SQL 写入不属于该约束入口。 +## 分片上传配额 + +`POST /resources/uploads` 在接收分片前预留整个文件的日额度和账号容量。文件大小必须严格小于 `resourceUploadMB`;同一账号最多 32 个未到期会话,跨后端实例共同计数。未满足配额的请求不会创建分片目录。分片和完成请求按会话串行执行;会话过期或已完成返回 404,重新创建会话后可整传重试。 + +完成上传时以数据库事务提交容量并消费预留,使用既有资源幂等标识。失败释放预留;跨 UTC 日期完成需要通过当天日额度检查。请求与响应字段不变。 + ## AutoDL 视频画面规格配置 渠道模型的 `capabilityConfig.video.resolutions` 与 `defaultResolution` 直接表示所选完整分辨率与默认值,始终生效;不再使用或返回 `customSizeEnabled`、`customScreenSpec`。AutoDL v2.3.0 未声明独立比例参数,因此 `ratios` 为空、`defaultRatio` 为空字符串,横竖与方形由完整分辨率标签表达。